diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/ColumnarSegmentCreationDataSource.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/ColumnarSegmentCreationDataSource.java
new file mode 100644
index 0000000000..0dee1fbdca
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/ColumnarSegmentCreationDataSource.java
@@ -0,0 +1,95 @@
+/**
+ * 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.segment.local.segment.creator;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.pinot.segment.local.segment.creator.impl.stats.ColumnarSegmentPreIndexStatsContainer;
+import org.apache.pinot.segment.spi.creator.SegmentCreationDataSource;
+import org.apache.pinot.segment.spi.creator.SegmentPreIndexStatsCollector;
+import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * SegmentCreationDataSource implementation that uses ColumnReaderFactory for columnar data access.
+ *
+ *
This data source enables columnar segment building by providing access to column readers
+ * instead of row-based record readers. It supports:
+ *
+ *
Columnar statistics collection
+ *
Column-wise data access through ColumnReaderFactory
+ *
Efficient handling of new columns with default values
+ *
Data type conversions during schema evolution
+ *
+ */
+public class ColumnarSegmentCreationDataSource implements SegmentCreationDataSource, Closeable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ColumnarSegmentCreationDataSource.class);
+
+ private final Map _columnReaders;
+
+ /**
+ * Create a ColumnarSegmentCreationDataSource.
+ *
+ * @param columnReaders Map of column name to ColumnReader instances
+ */
+ public ColumnarSegmentCreationDataSource(Map columnReaders) {
+ _columnReaders = columnReaders;
+ }
+
+ @Override
+ public SegmentPreIndexStatsCollector gatherStats(StatsCollectorConfig statsCollectorConfig) {
+ LOGGER.info("Gathering stats using columnar approach for {} columns", _columnReaders.size());
+
+ // Use columnar stats container that efficiently collects statistics column-wise
+ ColumnarSegmentPreIndexStatsContainer columnarSegmentPreIndexStatsContainer =
+ new ColumnarSegmentPreIndexStatsContainer(_columnReaders, statsCollectorConfig);
+ columnarSegmentPreIndexStatsContainer.init();
+ return columnarSegmentPreIndexStatsContainer;
+ }
+
+ @Override
+ public RecordReader getRecordReader() {
+ // For columnar building, we don't use RecordReader
+ // This method is required by the interface but should not be called in columnar mode
+ throw new UnsupportedOperationException(
+ "RecordReader not supported in columnar mode. Use getColumnReaders() instead.");
+ }
+
+ /**
+ * Get all column readers.
+ *
+ * @return Map of column name to ColumnReader
+ */
+ public Map getColumnReaders() {
+ return _columnReaders;
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ for (ColumnReader columnReader : _columnReaders.values()) {
+ columnReader.close();
+ }
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
index a55ef41dfd..eb59cbc7cc 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
@@ -68,6 +68,7 @@
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.FieldSpec.FieldType;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.ColumnReader;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.env.CommonsConfigurationUtils;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -421,6 +422,53 @@ public void indexColumn(String columnName, @Nullable int[] sortedDocIds, IndexSe
}
}
+ /**
+ * Index a column using a ColumnReader (column-major approach).
+ * This method processes the column values using the iterator pattern from ColumnReader.
+ *
+ * @param columnName Name of the column to index
+ * @param columnReader ColumnReader for the column data
+ * @throws IOException if indexing fails
+ */
+ @Override
+ public void indexColumn(String columnName, ColumnReader columnReader)
+ throws IOException {
+ Map, IndexCreator> creatorsByIndex = _creatorsByColAndIndex.get(columnName);
+ NullValueVectorCreator nullVec = _nullValueVectorCreatorMap.get(columnName);
+ FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
+ SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
+ Object defaultNullValue = fieldSpec.getDefaultNullValue();
+ Object reuseColumnValueToIndex;
+
+ // Reset column reader to start from beginning
+ columnReader.rewind();
+
+ int docId = 0;
+ while (columnReader.hasNext()) {
+ reuseColumnValueToIndex = columnReader.next();
+
+ // Handle null values
+ if (reuseColumnValueToIndex == null) {
+ if (nullVec != null) {
+ nullVec.setNull(docId);
+ }
+ reuseColumnValueToIndex = defaultNullValue;
+ }
+
+ try {
+ if (fieldSpec.isSingleValueField()) {
+ indexSingleValueRow(dictionaryCreator, reuseColumnValueToIndex, creatorsByIndex);
+ } else {
+ indexMultiValueRow(dictionaryCreator, (Object[]) reuseColumnValueToIndex, creatorsByIndex);
+ }
+ } catch (JsonParseException jpe) {
+ throw new ColumnJsonParserException(columnName, jpe);
+ }
+
+ docId++;
+ }
+ }
+
private void indexColumnValue(PinotSegmentColumnReader colReader,
Map, IndexCreator> creatorsByIndex, String columnName, FieldSpec fieldSpec,
SegmentDictionaryCreator dictionaryCreator, int sourceDocId, int onDiskDocPos,
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreator.java
index 2e4db2a6e6..b27f39ca69 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreator.java
@@ -332,6 +332,96 @@ public int indexOfSV(Object value) {
}
}
+ /**
+ * Get dictionary index for a primitive int value without boxing.
+ */
+ public int indexOfSV(int value) {
+ return _intValueToIndexMap.get(value);
+ }
+
+ /**
+ * Get dictionary index for a primitive long value without boxing.
+ */
+ public int indexOfSV(long value) {
+ return _longValueToIndexMap.get(value);
+ }
+
+ /**
+ * Get dictionary index for a primitive float value without boxing.
+ */
+ public int indexOfSV(float value) {
+ return _floatValueToIndexMap.get(value);
+ }
+
+ /**
+ * Get dictionary index for a primitive double value without boxing.
+ */
+ public int indexOfSV(double value) {
+ return _doubleValueToIndexMap.get(value);
+ }
+
+ /**
+ * Get dictionary index for a String value.
+ */
+ public int indexOfSV(String value) {
+ return _objectValueToIndexMap.getInt(value);
+ }
+
+ /**
+ * Get dictionary index for a byte array value.
+ */
+ public int indexOfSV(byte[] value) {
+ return _objectValueToIndexMap.getInt(new ByteArray(value));
+ }
+
+ public int[] indexOfMV(int[] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _intValueToIndexMap.get(values[i]);
+ }
+ return indexes;
+ }
+
+ public int[] indexOfMV(long[] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _longValueToIndexMap.get(values[i]);
+ }
+ return indexes;
+ }
+
+ public int[] indexOfMV(float[] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _floatValueToIndexMap.get(values[i]);
+ }
+ return indexes;
+ }
+
+ public int[] indexOfMV(double[] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _doubleValueToIndexMap.get(values[i]);
+ }
+ return indexes;
+ }
+
+ public int[] indexOfMV(String[] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _objectValueToIndexMap.getInt(values[i]);
+ }
+ return indexes;
+ }
+
+ public int[] indexOfMV(byte[][] values) {
+ int[] indexes = new int[values.length];
+ for (int i = 0; i < values.length; i++) {
+ indexes[i] = _objectValueToIndexMap.getInt(new ByteArray(values[i]));
+ }
+ return indexes;
+ }
+
public int[] indexOfMV(Object value) {
Object[] multiValues = (Object[]) value;
int[] indexes = new int[multiValues.length];
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
index 612b696e27..e33f1570d5 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
@@ -36,6 +36,7 @@
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.segment.local.realtime.converter.stats.RealtimeSegmentSegmentCreationDataSource;
+import org.apache.pinot.segment.local.segment.creator.ColumnarSegmentCreationDataSource;
import org.apache.pinot.segment.local.segment.creator.RecordReaderSegmentCreationDataSource;
import org.apache.pinot.segment.local.segment.creator.TransformPipeline;
import org.apache.pinot.segment.local.segment.index.converter.SegmentFormatConverterFactory;
@@ -75,6 +76,8 @@
import org.apache.pinot.spi.data.IngestionSchemaValidator;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.SchemaValidatorFactory;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.ColumnReaderFactory;
import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordReader;
@@ -84,8 +87,6 @@
import org.apache.pinot.spi.utils.ReadMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
-
/**
* Implementation of an index segment creator.
*/
@@ -94,7 +95,7 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
private static final Logger LOGGER = LoggerFactory.getLogger(SegmentIndexCreationDriverImpl.class);
private SegmentGeneratorConfig _config;
- private RecordReader _recordReader;
+ @Nullable private RecordReader _recordReader;
private SegmentPreIndexStatsContainer _segmentStats;
// NOTE: Use TreeMap so that the columns are ordered alphabetically
private TreeMap _indexCreationInfoMap;
@@ -102,7 +103,7 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
private SegmentIndexCreationInfo _segmentIndexCreationInfo;
private SegmentCreationDataSource _dataSource;
private Schema _dataSchema;
- private TransformPipeline _transformPipeline;
+ @Nullable private TransformPipeline _transformPipeline;
private IngestionSchemaValidator _ingestionSchemaValidator;
private int _totalDocs = 0;
private File _tempIndexDir;
@@ -162,31 +163,71 @@ public void init(SegmentGeneratorConfig config, RecordReader recordReader)
new TransformPipeline(config.getTableConfig(), config.getSchema()));
}
+ /**
+ * Initialize the driver for columnar segment building using a ColumnReaderFactory.
+ * This method sets up the driver to use column-wise input data access instead of row-wise.
+ *
+ * @param config Segment generator configuration
+ * @param columnReaderFactory Factory for creating column readers
+ * @throws Exception if initialization fails
+ */
+ public void init(SegmentGeneratorConfig config, ColumnReaderFactory columnReaderFactory)
+ throws Exception {
+ // Initialize the column reader factory with target schema
+ columnReaderFactory.init(config.getSchema());
+
+ // Get all column readers for the target schema
+ Map columnReaders = columnReaderFactory.getAllColumnReaders();
+
+ // Create columnar data source
+ ColumnarSegmentCreationDataSource columnarDataSource = new ColumnarSegmentCreationDataSource(columnReaders);
+
+ // Use the existing init method with columnar data source and no transform pipeline
+ init(config, columnarDataSource, null);
+
+ LOGGER.info("Initialized SegmentIndexCreationDriverImpl for columnar data source building with {} columns",
+ columnReaders.size());
+ }
+
public void init(SegmentGeneratorConfig config, SegmentCreationDataSource dataSource,
TransformPipeline transformPipeline)
throws Exception {
_config = config;
- _recordReader = dataSource.getRecordReader();
_dataSchema = config.getSchema();
_continueOnError = config.isContinueOnError();
+ String readerClassName = null;
+
+ // Handle columnar data sources differently
+ if (dataSource instanceof ColumnarSegmentCreationDataSource) {
+ // For columnar data sources, we don't have a record reader
+ _recordReader = null;
+ _transformPipeline = null; // No transform pipeline for columnar mode
+ _dataSource = dataSource;
+ } else {
+ // For record reader-based data sources
+ _recordReader = dataSource.getRecordReader();
- if (config.isFailOnEmptySegment()) {
- Preconditions.checkState(_recordReader.hasNext(), "No record in data source");
- }
- _transformPipeline = transformPipeline;
- // Use the same transform pipeline if the data source is backed by a record reader
- if (dataSource instanceof RecordReaderSegmentCreationDataSource) {
- ((RecordReaderSegmentCreationDataSource) dataSource).setTransformPipeline(transformPipeline);
- }
+ if (config.isFailOnEmptySegment()) {
+ Preconditions.checkState(_recordReader.hasNext(), "No record in data source");
+ }
+ _transformPipeline = transformPipeline;
+ // Use the same transform pipeline if the data source is backed by a record reader
+ if (dataSource instanceof RecordReaderSegmentCreationDataSource) {
+ ((RecordReaderSegmentCreationDataSource) dataSource).setTransformPipeline(transformPipeline);
+ }
- // Optimization for realtime segment conversion
- if (dataSource instanceof RealtimeSegmentSegmentCreationDataSource) {
- _config.setRealtimeConversion(true);
- _config.setConsumerDir(((RealtimeSegmentSegmentCreationDataSource) dataSource).getConsumerDir());
- }
+ // Optimization for realtime segment conversion
+ if (dataSource instanceof RealtimeSegmentSegmentCreationDataSource) {
+ _config.setRealtimeConversion(true);
+ _config.setConsumerDir(((RealtimeSegmentSegmentCreationDataSource) dataSource).getConsumerDir());
+ }
- // For stats collection
- _dataSource = dataSource;
+ // For stats collection
+ _dataSource = dataSource;
+
+ // Initialize common components
+ readerClassName = _recordReader.getClass().getName();
+ }
// Initialize index creation
_segmentIndexCreationInfo = new SegmentIndexCreationInfo();
@@ -202,7 +243,7 @@ public void init(SegmentGeneratorConfig config, SegmentCreationDataSource dataSo
}
_ingestionSchemaValidator =
- SchemaValidatorFactory.getSchemaValidator(_dataSchema, _recordReader.getClass().getName(),
+ SchemaValidatorFactory.getSchemaValidator(_dataSchema, readerClassName,
config.getInputFilePath());
// Create a temporary directory used in segment creation
@@ -230,6 +271,13 @@ private int[] getImmutableToMutableIdMap(@Nullable int[] sortedDocIds) {
@Override
public void build()
throws Exception {
+ // Check if we're using a columnar data source and switch to columnar mode
+ if (_dataSource instanceof ColumnarSegmentCreationDataSource) {
+ LOGGER.info("Detected columnar data source, using columnar building approach");
+ buildColumnar();
+ return;
+ }
+
// Count the number of documents and gather per-column statistics
LOGGER.debug("Start building StatsCollector!");
collectStatsAndIndexCreationInfo();
@@ -586,4 +634,83 @@ public IngestionSchemaValidator getIngestionSchemaValidator() {
public SegmentPreIndexStatsContainer getSegmentStats() {
return _segmentStats;
}
+
+ /**
+ * Build segment using columnar approach.
+ * This method builds the segment by processing data column-wise instead of row-wise.
+ * Following is not supported:
+ *
record transformation
+ *
sorted column change wrt to input data
+ *
+ *
Initialize the driver using {@link #init(SegmentGeneratorConfig, ColumnReaderFactory)}
+ *
+ * @throws Exception if segment building fails
+ */
+ private void buildColumnar()
+ throws Exception {
+ if (!(_dataSource instanceof ColumnarSegmentCreationDataSource)) {
+ throw new IllegalStateException("buildColumnar() can only be called after initColumnar()");
+ }
+
+ ColumnarSegmentCreationDataSource columnarDataSource = (ColumnarSegmentCreationDataSource) _dataSource;
+
+ try {
+ Map columnReaders = columnarDataSource.getColumnReaders();
+
+ LOGGER.info("Starting columnar segment building for {} columns", columnReaders.size());
+
+ // Reuse existing stats collection and index creation info logic
+ LOGGER.debug("Start building StatsCollector!");
+ collectStatsAndIndexCreationInfo();
+ LOGGER.info("Finished building StatsCollector!");
+ LOGGER.info("Collected stats for {} documents", _totalDocs);
+
+ if (_totalDocs == 0) {
+ LOGGER.warn("No documents found in data source");
+ handlePostCreation();
+ return;
+ }
+
+ // Initialize the index creation using the per-column statistics information
+ _indexCreator.init(_config, _segmentIndexCreationInfo, _indexCreationInfoMap, _dataSchema, _tempIndexDir, null);
+
+ // Build the indexes column-wise (true column-major approach)
+ LOGGER.info("Start building Index using columnar approach");
+ long indexStartTime = System.nanoTime();
+
+ TreeSet columns = _dataSchema.getPhysicalColumnNames();
+ for (String columnName : columns) {
+ LOGGER.debug("Indexing column: {}", columnName);
+ ColumnReader columnReader = columnReaders.get(columnName);
+ if (columnReader == null) {
+ throw new IllegalStateException("No column reader found for column: " + columnName);
+ }
+
+ // Index each column independently using true column-major approach
+ // This is similar to how buildByColumn works but uses ColumnReader instead of IndexSegment
+ _indexCreator.indexColumn(columnName, columnReader);
+ }
+
+ _totalIndexTimeNs = System.nanoTime() - indexStartTime;
+
+ LOGGER.info("Finished indexing using columnar approach in IndexCreator!");
+ handlePostCreation();
+ } catch (Exception e) {
+ try {
+ _indexCreator.close();
+ } catch (Exception closeException) {
+ LOGGER.error("Error closing index creator", closeException);
+ // Add the close exception as suppressed to preserve both exceptions
+ e.addSuppressed(closeException);
+ }
+ throw e;
+ } finally {
+ // Always close the columnar data source to prevent resource leaks
+ try {
+ columnarDataSource.close();
+ } catch (Exception closeException) {
+ LOGGER.error("Error closing columnar data source", closeException);
+ }
+ }
+ }
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollector.java
index f5aa512c43..f4f3f302fb 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollector.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollector.java
@@ -99,6 +99,40 @@ public boolean isSorted() {
*/
public abstract void collect(Object entry);
+ // Collects statistics for primitive values
+ // Default implementation boxes the value for backward compatibility.
+ public void collect(int value) {
+ collect((Object) value);
+ }
+
+ public void collect(long value) {
+ collect((Object) value);
+ }
+
+ public void collect(float value) {
+ collect((Object) value);
+ }
+
+ public void collect(double value) {
+ collect((Object) value);
+ }
+
+ public void collect(int[] values) {
+ collect((Object) values);
+ }
+
+ public void collect(long[] values) {
+ collect((Object) values);
+ }
+
+ public void collect(float[] values) {
+ collect((Object) values);
+ }
+
+ public void collect(double[] values) {
+ collect((Object) values);
+ }
+
public int getLengthOfShortestElement() {
return -1;
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnarSegmentPreIndexStatsContainer.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnarSegmentPreIndexStatsContainer.java
new file mode 100644
index 0000000000..88d829ae06
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnarSegmentPreIndexStatsContainer.java
@@ -0,0 +1,178 @@
+/**
+ * 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.segment.local.segment.creator.impl.stats;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.segment.spi.creator.ColumnStatistics;
+import org.apache.pinot.segment.spi.creator.SegmentPreIndexStatsCollector;
+import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Stats container that efficiently collects statistics from columnar data using ColumnReader instances.
+ *
+ *
This implementation collects statistics by iterating column-wise instead of row-wise,
+ * which is more efficient for columnar data sources. It supports:
+ *
+ *
Column-wise statistics collection
+ *
Existing columns from source data
+ *
New columns with default values
+ *
Data type conversions during schema evolution
+ *
+ *
+ *
The statistics are collected using the same underlying collectors as the row-based approach
+ * (SegmentPreIndexStatsCollectorImpl) but with more efficient column-wise iteration.
+ */
+public class ColumnarSegmentPreIndexStatsContainer implements SegmentPreIndexStatsCollector {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ColumnarSegmentPreIndexStatsContainer.class);
+
+ private final Map _columnReaders;
+ private final StatsCollectorConfig _statsCollectorConfig;
+ private final Schema _targetSchema;
+ private final Map _columnStatsCollectorMap;
+ private int _totalDocCount;
+
+ /**
+ * Create a ColumnarSegmentPreIndexStatsContainer.
+ *
+ * @param columnReaders Map of column name to ColumnReader instances
+ * @param statsCollectorConfig Configuration for statistics collection
+ */
+ public ColumnarSegmentPreIndexStatsContainer(Map columnReaders,
+ StatsCollectorConfig statsCollectorConfig) {
+ _columnReaders = columnReaders;
+ _statsCollectorConfig = statsCollectorConfig;
+ _targetSchema = statsCollectorConfig.getSchema();
+ _columnStatsCollectorMap = new HashMap<>(columnReaders.size());
+ _totalDocCount = -1; // indicates unset
+ }
+
+ /**
+ * Initialize stats collectors for all columns in the target schema.
+ */
+ private void initializeStatsCollectors() {
+ for (FieldSpec fieldSpec : _targetSchema.getAllFieldSpecs()) {
+ if (fieldSpec.isVirtualColumn()) {
+ continue;
+ }
+
+ String columnName = fieldSpec.getName();
+ AbstractColumnStatisticsCollector collector =
+ StatsCollectorUtil.createStatsCollector(columnName, fieldSpec, _statsCollectorConfig);
+ _columnStatsCollectorMap.put(columnName, collector);
+ }
+ }
+
+ /**
+ * Collect stats by iterating column-wise using the provided ColumnReader instances.
+ */
+ private void collectColumnStats() {
+ LOGGER.info("Collecting stats for {} columns using column-wise iteration", _columnReaders.size());
+
+ for (String columnName : _columnStatsCollectorMap.keySet()) {
+ AbstractColumnStatisticsCollector statsCollector = _columnStatsCollectorMap.get(columnName);
+ ColumnReader columnReader = _columnReaders.get(columnName);
+
+ if (columnReader == null) {
+ throw new RuntimeException("Column reader for column " + columnName + " not found");
+ }
+
+ LOGGER.debug("Collecting stats for column: {}", columnName);
+ collectStatsFromColumnReader(columnName, columnReader, statsCollector);
+
+ // Seal the stats collector
+ statsCollector.seal();
+ }
+ }
+
+ /**
+ * Collect stats from a column reader by iterating over all values using the iterator pattern.
+ */
+ private void collectStatsFromColumnReader(String columnName, ColumnReader columnReader,
+ AbstractColumnStatisticsCollector statsCollector) {
+ try {
+ // Reset the column reader to start from the beginning
+ columnReader.rewind();
+
+ int docCount = 0;
+ while (columnReader.hasNext()) {
+ Object value = columnReader.next();
+ statsCollector.collect(value);
+ docCount++;
+ }
+
+ // if totalDocCount is unset then set total doc count from the first column
+ if (_totalDocCount == -1) {
+ _totalDocCount = docCount;
+ } else if (_totalDocCount != docCount) {
+ // all columns should have same count
+ LOGGER.warn("Column {} has {} documents, but expected {} documents",
+ columnName, docCount, _totalDocCount);
+ throw new RuntimeException("Columns have inconsistent document counts");
+ }
+ } catch (IOException e) {
+ LOGGER.error("Failed to collect stats for column: {}", columnName, e);
+ throw new RuntimeException("Failed to collect stats for column: " + columnName, e);
+ }
+ }
+
+ @Override
+ public ColumnStatistics getColumnProfileFor(String column) {
+ return _columnStatsCollectorMap.get(column);
+ }
+
+ @Override
+ public int getTotalDocCount() {
+ return _totalDocCount;
+ }
+
+ @Override
+ public void init() {
+ initializeStatsCollectors();
+ collectColumnStats();
+ }
+
+ @Override
+ public void build() {
+ // Stats are already collected in constructor
+ }
+
+ @Override
+ public void logStats() {
+ LOGGER.info("Columnar segment stats collection completed for {} columns with {} documents",
+ _columnStatsCollectorMap.size(), _totalDocCount);
+ }
+
+ @Override
+ public void collectRow(GenericRow row) {
+ // This method is not used in columnar stats collection as we iterate column-wise
+ // instead of row-wise. The stats are collected in the constructor by iterating
+ // over each column using ColumnReader instances.
+ throw new UnsupportedOperationException(
+ "collectRow is not supported in columnar stats collection. Stats are collected column-wise.");
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java
index 9edd89b027..77b34b0459 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java
@@ -40,32 +40,38 @@ public void collect(Object entry) {
if (entry instanceof Object[]) {
Object[] values = (Object[]) entry;
- for (Object obj : values) {
- double value = (double) obj;
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values);
+ collectMultiValue(values.length, i -> (double) values[i]);
} else if (entry instanceof double[]) {
- double[] values = (double[]) entry;
- for (double value : values) {
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values.length);
+ collect((double[]) entry);
} else {
- double value = (double) entry;
- addressSorted(value);
- if (_values.add(value)) {
- if (isPartitionEnabled()) {
- updatePartition(Double.toString(value));
- }
+ collect((double) entry);
+ }
+ }
+
+ @Override
+ public void collect(double value) {
+ assert !_sealed;
+ addressSorted(value);
+ if (_values.add(value)) {
+ if (isPartitionEnabled()) {
+ updatePartition(Double.toString(value));
}
+ }
+ _totalNumberOfEntries++;
+ }
+
+ @Override
+ public void collect(double[] values) {
+ assert !_sealed;
+ collectMultiValue(values.length, i -> values[i]);
+ }
- _totalNumberOfEntries++;
+ private void collectMultiValue(int length, java.util.function.IntFunction valueGetter) {
+ for (int i = 0; i < length; i++) {
+ _values.add(valueGetter.apply(i));
}
+ _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, length);
+ updateTotalNumberOfEntries(length);
}
private void addressSorted(double entry) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java
index 4be248124e..aa60f4d5ba 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java
@@ -40,32 +40,38 @@ public void collect(Object entry) {
if (entry instanceof Object[]) {
Object[] values = (Object[]) entry;
- for (Object obj : values) {
- float value = (float) obj;
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values);
+ collectMultiValue(values.length, i -> (float) values[i]);
} else if (entry instanceof float[]) {
- float[] values = (float[]) entry;
- for (float value : values) {
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values.length);
+ collect((float[]) entry);
} else {
- float value = (float) entry;
- addressSorted(value);
- if (_values.add(value)) {
- if (isPartitionEnabled()) {
- updatePartition(Float.toString(value));
- }
+ collect((float) entry);
+ }
+ }
+
+ @Override
+ public void collect(float value) {
+ assert !_sealed;
+ addressSorted(value);
+ if (_values.add(value)) {
+ if (isPartitionEnabled()) {
+ updatePartition(Float.toString(value));
}
+ }
+ _totalNumberOfEntries++;
+ }
+
+ @Override
+ public void collect(float[] values) {
+ assert !_sealed;
+ collectMultiValue(values.length, i -> values[i]);
+ }
- _totalNumberOfEntries++;
+ private void collectMultiValue(int length, java.util.function.IntFunction valueGetter) {
+ for (int i = 0; i < length; i++) {
+ _values.add(valueGetter.apply(i));
}
+ _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, length);
+ updateTotalNumberOfEntries(length);
}
private void addressSorted(float entry) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java
index 29678f6805..74882f25ff 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java
@@ -40,32 +40,38 @@ public void collect(Object entry) {
if (entry instanceof Object[]) {
Object[] values = (Object[]) entry;
- for (Object obj : values) {
- int value = (int) obj;
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values);
+ collectMultiValue(values.length, i -> (int) values[i]);
} else if (entry instanceof int[]) {
- int[] values = (int[]) entry;
- for (int value : values) {
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values.length);
+ collect((int[]) entry);
} else {
- int value = (int) entry;
- addressSorted(value);
- if (_values.add(value)) {
- if (isPartitionEnabled()) {
- updatePartition(Integer.toString(value));
- }
+ collect((int) entry);
+ }
+ }
+
+ @Override
+ public void collect(int value) {
+ assert !_sealed;
+ addressSorted(value);
+ if (_values.add(value)) {
+ if (isPartitionEnabled()) {
+ updatePartition(Integer.toString(value));
}
+ }
+ _totalNumberOfEntries++;
+ }
+
+ @Override
+ public void collect(int[] values) {
+ assert !_sealed;
+ collectMultiValue(values.length, i -> values[i]);
+ }
- _totalNumberOfEntries++;
+ private void collectMultiValue(int length, java.util.function.IntFunction valueGetter) {
+ for (int i = 0; i < length; i++) {
+ _values.add(valueGetter.apply(i));
}
+ _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, length);
+ updateTotalNumberOfEntries(length);
}
private void addressSorted(int entry) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java
index 3d4c63d03b..c60efc5bd9 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java
@@ -40,32 +40,38 @@ public void collect(Object entry) {
if (entry instanceof Object[]) {
Object[] values = (Object[]) entry;
- for (Object obj : values) {
- long value = (long) obj;
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values);
+ collectMultiValue(values.length, i -> (long) values[i]);
} else if (entry instanceof long[]) {
- long[] values = (long[]) entry;
- for (long value : values) {
- _values.add(value);
- }
-
- _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, values.length);
- updateTotalNumberOfEntries(values.length);
+ collect((long[]) entry);
} else {
- long value = (long) entry;
- addressSorted(value);
- if (_values.add(value)) {
- if (isPartitionEnabled()) {
- updatePartition(Long.toString(value));
- }
+ collect((long) entry);
+ }
+ }
+
+ @Override
+ public void collect(long value) {
+ assert !_sealed;
+ addressSorted(value);
+ if (_values.add(value)) {
+ if (isPartitionEnabled()) {
+ updatePartition(Long.toString(value));
}
+ }
+ _totalNumberOfEntries++;
+ }
+
+ @Override
+ public void collect(long[] values) {
+ assert !_sealed;
+ collectMultiValue(values.length, i -> values[i]);
+ }
- _totalNumberOfEntries++;
+ private void collectMultiValue(int length, java.util.function.IntFunction valueGetter) {
+ for (int i = 0; i < length; i++) {
+ _values.add(valueGetter.apply(i));
}
+ _maxNumberOfMultiValues = Math.max(_maxNumberOfMultiValues, length);
+ updateTotalNumberOfEntries(length);
}
private void addressSorted(long entry) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/SegmentPreIndexStatsCollectorImpl.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/SegmentPreIndexStatsCollectorImpl.java
index 45c75e0c49..0a680550a9 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/SegmentPreIndexStatsCollectorImpl.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/SegmentPreIndexStatsCollectorImpl.java
@@ -48,35 +48,9 @@ public void init() {
Schema dataSchema = _statsCollectorConfig.getSchema();
for (FieldSpec fieldSpec : dataSchema.getAllFieldSpecs()) {
String column = fieldSpec.getName();
- switch (fieldSpec.getDataType().getStoredType()) {
- case INT:
- _columnStatsCollectorMap.put(column, new IntColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case LONG:
- _columnStatsCollectorMap.put(column, new LongColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case FLOAT:
- _columnStatsCollectorMap.put(column, new FloatColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case DOUBLE:
- _columnStatsCollectorMap.put(column, new DoubleColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case BIG_DECIMAL:
- _columnStatsCollectorMap.put(column,
- new BigDecimalColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case STRING:
- _columnStatsCollectorMap.put(column, new StringColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case BYTES:
- _columnStatsCollectorMap.put(column, new BytesColumnPredIndexStatsCollector(column, _statsCollectorConfig));
- break;
- case MAP:
- _columnStatsCollectorMap.put(column, new MapColumnPreIndexStatsCollector(column, _statsCollectorConfig));
- break;
- default:
- throw new IllegalStateException("Unsupported data type: " + fieldSpec.getDataType());
- }
+ AbstractColumnStatisticsCollector collector =
+ StatsCollectorUtil.createStatsCollector(column, fieldSpec, _statsCollectorConfig);
+ _columnStatsCollectorMap.put(column, collector);
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/StatsCollectorUtil.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/StatsCollectorUtil.java
new file mode 100644
index 0000000000..469184c3f4
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/StatsCollectorUtil.java
@@ -0,0 +1,65 @@
+/**
+ * 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.segment.local.segment.creator.impl.stats;
+
+import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.spi.data.FieldSpec;
+
+
+/**
+ * Utility class for creating column statistics collectors.
+ */
+public final class StatsCollectorUtil {
+
+ private StatsCollectorUtil() {
+ // Utility class
+ }
+
+ /**
+ * Create a statistics collector for the given column based on its data type.
+ *
+ * @param columnName Name of the column
+ * @param fieldSpec Field specification for the column
+ * @param statsCollectorConfig Stats collector configuration
+ * @return AbstractColumnStatisticsCollector for the column
+ */
+ public static AbstractColumnStatisticsCollector createStatsCollector(String columnName, FieldSpec fieldSpec,
+ StatsCollectorConfig statsCollectorConfig) {
+ switch (fieldSpec.getDataType().getStoredType()) {
+ case INT:
+ return new IntColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case LONG:
+ return new LongColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case FLOAT:
+ return new FloatColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case DOUBLE:
+ return new DoubleColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case BIG_DECIMAL:
+ return new BigDecimalColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case STRING:
+ return new StringColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ case BYTES:
+ return new BytesColumnPredIndexStatsCollector(columnName, statsCollectorConfig);
+ case MAP:
+ return new MapColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
+ default:
+ throw new IllegalStateException("Unsupported data type: " + fieldSpec.getDataType());
+ }
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReader.java
new file mode 100644
index 0000000000..4392fe9cd0
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReader.java
@@ -0,0 +1,410 @@
+/**
+ * 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.segment.local.segment.readers;
+
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.MultiValueResult;
+
+
+/**
+ * ColumnReader implementation that returns default values for new columns.
+ *
+ *
This reader is used when a column exists in the target schema but not in the source data.
+ * It returns the default null value for the field spec for all document IDs.
+ */
+public class DefaultValueColumnReader implements ColumnReader {
+
+ private final String _columnName;
+ private final int _numDocs;
+ private final Object _defaultValue;
+ private final FieldSpec _fieldSpec;
+ private final FieldSpec.DataType _dataType;
+
+ // Pre-computed multi-value arrays for reuse
+ private int[] _defaultIntMV;
+ private long[] _defaultLongMV;
+ private float[] _defaultFloatMV;
+ private double[] _defaultDoubleMV;
+ private String[] _defaultStringMV;
+ private byte[][] _defaultBytesMV;
+
+ private int _currentIndex;
+
+ /**
+ * Create a DefaultValueColumnReader for a new column.
+ *
+ * @param columnName Name of the new column
+ * @param numDocs Total number of documents
+ * @param fieldSpec Field specification for the new column
+ */
+ public DefaultValueColumnReader(String columnName, int numDocs, FieldSpec fieldSpec) {
+ _columnName = columnName;
+ _numDocs = numDocs;
+ _currentIndex = 0;
+ _fieldSpec = fieldSpec;
+ _dataType = fieldSpec.getDataType();
+
+ // For multi-value fields, wrap the default value in an array
+ Object defaultNullValue = fieldSpec.getDefaultNullValue();
+ if (fieldSpec.isSingleValueField()) {
+ _defaultValue = defaultNullValue;
+ } else {
+ _defaultValue = new Object[]{defaultNullValue};
+ // Pre-compute typed arrays for multi-value fields to avoid repeated allocations
+ Object[] defaultArray = (Object[]) _defaultValue;
+ switch (_dataType) {
+ case INT:
+ _defaultIntMV = new int[defaultArray.length];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultIntMV[i] = ((Number) defaultArray[i]).intValue();
+ }
+ break;
+ case LONG:
+ _defaultLongMV = new long[defaultArray.length];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultLongMV[i] = ((Number) defaultArray[i]).longValue();
+ }
+ break;
+ case FLOAT:
+ _defaultFloatMV = new float[defaultArray.length];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultFloatMV[i] = ((Number) defaultArray[i]).floatValue();
+ }
+ break;
+ case DOUBLE:
+ _defaultDoubleMV = new double[defaultArray.length];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultDoubleMV[i] = ((Number) defaultArray[i]).doubleValue();
+ }
+ break;
+ case STRING:
+ _defaultStringMV = new String[defaultArray.length];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultStringMV[i] = (String) defaultArray[i];
+ }
+ break;
+ case BYTES:
+ _defaultBytesMV = new byte[defaultArray.length][];
+ for (int i = 0; i < defaultArray.length; i++) {
+ _defaultBytesMV[i] = (byte[]) defaultArray[i];
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return _currentIndex < _numDocs;
+ }
+
+ @Override
+ @Nullable
+ public Object next() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return _defaultValue;
+ }
+
+ @Override
+ public void rewind() {
+ _currentIndex = 0;
+ }
+
+ @Override
+ public boolean isNextNull() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ return _defaultValue == null;
+ }
+
+ @Override
+ public void skipNext() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return _fieldSpec.isSingleValueField();
+ }
+
+ @Override
+ public boolean isInt() {
+ return _dataType == FieldSpec.DataType.INT;
+ }
+
+ @Override
+ public boolean isLong() {
+ return _dataType == FieldSpec.DataType.LONG;
+ }
+
+ @Override
+ public boolean isFloat() {
+ return _dataType == FieldSpec.DataType.FLOAT;
+ }
+
+ @Override
+ public boolean isDouble() {
+ return _dataType == FieldSpec.DataType.DOUBLE;
+ }
+
+ @Override
+ public boolean isString() {
+ return _dataType == FieldSpec.DataType.STRING;
+ }
+
+ @Override
+ public boolean isBytes() {
+ return _dataType == FieldSpec.DataType.BYTES;
+ }
+
+ @Override
+ public int nextInt() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return ((Number) _defaultValue).intValue();
+ }
+
+ @Override
+ public long nextLong() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return ((Number) _defaultValue).longValue();
+ }
+
+ @Override
+ public float nextFloat() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return ((Number) _defaultValue).floatValue();
+ }
+
+ @Override
+ public double nextDouble() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return ((Number) _defaultValue).doubleValue();
+ }
+
+ @Override
+ public String nextString() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return (String) _defaultValue;
+ }
+
+ @Override
+ public byte[] nextBytes() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return (byte[]) _defaultValue;
+ }
+
+ @Override
+ public MultiValueResult nextIntMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return MultiValueResult.of(_defaultIntMV, null);
+ }
+
+ @Override
+ public MultiValueResult nextLongMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return MultiValueResult.of(_defaultLongMV, null);
+ }
+
+ @Override
+ public MultiValueResult nextFloatMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return MultiValueResult.of(_defaultFloatMV, null);
+ }
+
+ @Override
+ public MultiValueResult nextDoubleMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return MultiValueResult.of(_defaultDoubleMV, null);
+ }
+
+ @Override
+ public String[] nextStringMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return _defaultStringMV;
+ }
+
+ @Override
+ public byte[][] nextBytesMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ return _defaultBytesMV;
+ }
+
+ @Override
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ @Override
+ public int getTotalDocs() {
+ return _numDocs;
+ }
+
+ @Override
+ public boolean isNull(int docId) {
+ validateDocId(docId);
+ return _defaultValue == null;
+ }
+
+ // Single-value accessors
+
+ @Override
+ public int getInt(int docId) {
+ validateDocId(docId);
+ return ((Number) _defaultValue).intValue();
+ }
+
+ @Override
+ public long getLong(int docId) {
+ validateDocId(docId);
+ return ((Number) _defaultValue).longValue();
+ }
+
+ @Override
+ public float getFloat(int docId) {
+ validateDocId(docId);
+ return ((Number) _defaultValue).floatValue();
+ }
+
+ @Override
+ public double getDouble(int docId) {
+ validateDocId(docId);
+ return ((Number) _defaultValue).doubleValue();
+ }
+
+ @Override
+ public String getString(int docId) {
+ validateDocId(docId);
+ return (String) _defaultValue;
+ }
+
+ @Override
+ public byte[] getBytes(int docId) {
+ validateDocId(docId);
+ return (byte[]) _defaultValue;
+ }
+
+ @Override
+ public Object getValue(int docId) {
+ validateDocId(docId);
+ return _defaultValue;
+ }
+
+ // Multi-value accessors
+
+ @Override
+ public MultiValueResult getIntMV(int docId) {
+ validateDocId(docId);
+ return MultiValueResult.of(_defaultIntMV, null);
+ }
+
+ @Override
+ public MultiValueResult getLongMV(int docId) {
+ validateDocId(docId);
+ return MultiValueResult.of(_defaultLongMV, null);
+ }
+
+ @Override
+ public MultiValueResult getFloatMV(int docId) {
+ validateDocId(docId);
+ return MultiValueResult.of(_defaultFloatMV, null);
+ }
+
+ @Override
+ public MultiValueResult getDoubleMV(int docId) {
+ validateDocId(docId);
+ return MultiValueResult.of(_defaultDoubleMV, null);
+ }
+
+ @Override
+ public String[] getStringMV(int docId) {
+ validateDocId(docId);
+ return _defaultStringMV;
+ }
+
+ @Override
+ public byte[][] getBytesMV(int docId) {
+ validateDocId(docId);
+ return _defaultBytesMV;
+ }
+
+ /**
+ * Validate that the document ID is within valid range.
+ *
+ * @param docId Document ID to validate
+ * @throws IndexOutOfBoundsException if docId is out of range
+ */
+ private void validateDocId(int docId) {
+ if (docId < 0 || docId >= _numDocs) {
+ throw new IndexOutOfBoundsException(
+ "docId " + docId + " is out of range. Valid range is 0 to " + (_numDocs - 1));
+ }
+ }
+
+ @Override
+ public void close() {
+ // No resources to close
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReader.java
index 4a691e4fc4..4e62b6e5af 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReader.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReader.java
@@ -177,6 +177,124 @@ public boolean isNull(int docId) {
return _nullValueVectorReader != null && _nullValueVectorReader.isNull(docId);
}
+ // Single-value accessors
+
+ public int getInt(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getIntValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getInt(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public long getLong(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getLongValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getLong(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public float getFloat(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getFloatValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getFloat(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public double getDouble(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getDoubleValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getDouble(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public String getString(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getStringValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getString(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public byte[] getBytes(int docId) {
+ if (_dictionary != null) {
+ return _dictionary.getBytesValue(_forwardIndexReader.getDictId(docId, _forwardIndexReaderContext));
+ } else {
+ return _forwardIndexReader.getBytes(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ // Multi-value accessors
+
+ public int[] getIntMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ int[] values = new int[numValues];
+ _dictionary.readIntValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getIntMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public long[] getLongMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ long[] values = new long[numValues];
+ _dictionary.readLongValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getLongMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public float[] getFloatMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ float[] values = new float[numValues];
+ _dictionary.readFloatValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getFloatMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public double[] getDoubleMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ double[] values = new double[numValues];
+ _dictionary.readDoubleValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getDoubleMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public String[] getStringMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ String[] values = new String[numValues];
+ _dictionary.readStringValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getStringMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
+ public byte[][] getBytesMV(int docId) {
+ if (_dictionary != null) {
+ int numValues = _forwardIndexReader.getDictIdMV(docId, _dictIdBuffer, _forwardIndexReaderContext);
+ byte[][] values = new byte[numValues][];
+ _dictionary.readBytesValues(_dictIdBuffer, numValues, values);
+ return values;
+ } else {
+ return _forwardIndexReader.getBytesMV(docId, _forwardIndexReaderContext);
+ }
+ }
+
@Override
public void close()
throws IOException {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderFactory.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderFactory.java
new file mode 100644
index 0000000000..b6eb206c31
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderFactory.java
@@ -0,0 +1,202 @@
+/**
+ * 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.segment.local.segment.readers;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.ColumnReaderFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * ColumnReaderFactory implementation for immutable Pinot segments.
+ *
+ *
This factory creates ColumnReader instances for reading data from Pinot segments
+ * in a columnar fashion. It handles:
+ *
+ *
Creating readers for existing columns in the segment
+ *
Creating default value readers for new columns
+ *
Resource management for all created readers
+ *
+ */
+public class PinotSegmentColumnReaderFactory implements ColumnReaderFactory {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PinotSegmentColumnReaderFactory.class);
+
+ private final IndexSegment _indexSegment;
+ private Schema _targetSchema;
+ private Set _colsToRead;
+ private Map _columnReaders;
+ private final boolean _initializeDefaultValueReaders;
+
+ /**
+ * Create a PinotSegmentColumnReaderFactory.
+ *
+ * @param indexSegment Source segment to read from
+ */
+ public PinotSegmentColumnReaderFactory(IndexSegment indexSegment) {
+ this(indexSegment, true);
+ }
+
+ /**
+ * Create a PinotSegmentColumnReaderFactory.
+ *
+ * @param indexSegment Source segment to read from
+ * @param initializeDefaultValueReaders Whether to initialize default value readers for missing columns
+ * TODO - Ideally this factory shouldn't initialize default value readers.
+ * The clients of this factory should decide whether to create default value readers or not.
+ * This parameter is kept for backward compatibility and will be removed in future.
+ */
+ public PinotSegmentColumnReaderFactory(IndexSegment indexSegment, boolean initializeDefaultValueReaders) {
+ _indexSegment = indexSegment;
+ _columnReaders = new HashMap<>();
+ _initializeDefaultValueReaders = initializeDefaultValueReaders;
+ }
+
+ @Override
+ public void init(Schema targetSchema)
+ throws IOException {
+ init(targetSchema, targetSchema.getPhysicalColumnNames());
+ }
+
+ @Override
+ public void init(Schema targetSchema, Set colsToRead)
+ throws IOException {
+ _targetSchema = targetSchema;
+ _colsToRead = colsToRead;
+ _columnReaders = initializeAllColumnReaders();
+ LOGGER.info("Initialized PinotSegmentColumnReaderFactory with target schema containing {} columns",
+ targetSchema.getPhysicalColumnNames().size());
+ }
+
+ @Override
+ public Set getAvailableColumns() {
+ return _indexSegment.getPhysicalColumnNames();
+ }
+
+ /**
+ * Get the total number of documents/rows in the data source.
+ *
+ * @return Total number of documents
+ */
+ public int getNumDocs() {
+ return _indexSegment.getSegmentMetadata().getTotalDocs();
+ }
+
+ @Override
+ public ColumnReader getColumnReader(String columnName) {
+ if (_targetSchema == null) {
+ throw new IllegalStateException("Factory not initialized. Call init() first.");
+ }
+
+ return _columnReaders.get(columnName);
+ }
+
+ /**
+ * Internal method to create a column reader for the specified column.
+ * This method is called during initialization to create all readers.
+ */
+ @Nullable
+ private ColumnReader createColumnReader(String columnName, FieldSpec targetFieldSpec) {
+ if (targetFieldSpec.isVirtualColumn()) {
+ throw new IllegalStateException("Target field spec is a virtual column.");
+ }
+
+ ColumnReader columnReader = null;
+
+ if (hasColumn(columnName)) {
+ // Column exists in source segment - create a segment column reader
+ LOGGER.debug("Creating segment column reader for existing column: {}", columnName);
+ columnReader = new PinotSegmentColumnReaderImpl(_indexSegment, columnName);
+ } else {
+ if (_initializeDefaultValueReaders) {
+ // New column - create a default value reader
+ LOGGER.debug("Creating default value reader for new column: {}", columnName);
+ columnReader = new DefaultValueColumnReader(columnName, getNumDocs(), targetFieldSpec);
+ }
+ }
+
+ return columnReader;
+ }
+
+ @Override
+ public Map getAllColumnReaders()
+ throws IOException {
+ return _columnReaders;
+ }
+
+ /**
+ * Internal method to initialize all column readers during factory initialization.
+ */
+ private Map initializeAllColumnReaders() {
+ if (_targetSchema == null) {
+ throw new IllegalStateException("Factory not initialized. Call init() first.");
+ }
+
+ Map allReaders = new HashMap<>();
+
+ // Create readers for all columns in the target schema
+ for (String columnName : _colsToRead) {
+ FieldSpec fieldSpec = _targetSchema.getFieldSpecFor(columnName);
+ if (fieldSpec.isVirtualColumn()) {
+ continue;
+ }
+ ColumnReader reader = createColumnReader(columnName, fieldSpec);
+ if (reader != null) {
+ allReaders.put(columnName, reader);
+ }
+ }
+
+ return allReaders;
+ }
+
+ /**
+ * Check if the specified column exists in the source data.
+ *
+ * @param columnName Column name to check
+ * @return true if the column exists in source, false otherwise
+ */
+ public boolean hasColumn(String columnName) {
+ return _indexSegment.getPhysicalColumnNames().contains(columnName);
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ LOGGER.debug("Closing PinotSegmentColumnReaderFactory and {} column readers", _columnReaders.size());
+
+ // Close all created column readers
+ for (ColumnReader reader : _columnReaders.values()) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ LOGGER.warn("Error closing column reader for column: {}", reader.getColumnName(), e);
+ }
+ }
+
+ _columnReaders.clear();
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderImpl.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderImpl.java
new file mode 100644
index 0000000000..b34a0e53ef
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnReaderImpl.java
@@ -0,0 +1,382 @@
+/**
+ * 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.segment.local.segment.readers;
+
+import java.io.IOException;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.data.readers.MultiValueResult;
+
+
+/**
+ * Implementation of ColumnReader for Pinot segments.
+ *
+ *
This class wraps the existing PinotSegmentColumnReader and provides the ColumnReader interface
+ * for columnar segment building. It handles:
+ *
+ *
Reading column values from Pinot segments
+ *
Resource cleanup
+ *
+ */
+public class PinotSegmentColumnReaderImpl implements ColumnReader {
+ private final PinotSegmentColumnReader _segmentColumnReader;
+ private final String _columnName;
+ private final int _numDocs;
+ private final FieldSpec.DataType _dataType;
+
+ private int _currentIndex;
+
+ // Reusable variables to avoid garbage collection on every next() call
+ private Object _reuseValue;
+
+ /**
+ * Create a PinotSegmentColumnReaderImpl for an existing column in the segment.
+ *
+ * @param indexSegment Source segment to read from
+ * @param columnName Name of the column
+ */
+ public PinotSegmentColumnReaderImpl(IndexSegment indexSegment, String columnName) {
+ _segmentColumnReader = new PinotSegmentColumnReader(indexSegment, columnName);
+ _columnName = columnName;
+ _numDocs = indexSegment.getSegmentMetadata().getTotalDocs();
+ _currentIndex = 0;
+
+ // Get the data type from the schema
+ FieldSpec fieldSpec = indexSegment.getSegmentMetadata().getSchema().getFieldSpecFor(columnName);
+ assert fieldSpec != null : "FieldSpec should not be null for column: " + columnName;
+ _dataType = fieldSpec.getDataType();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return _currentIndex < _numDocs;
+ }
+
+ @Override
+ @Nullable
+ public Object next()
+ throws IOException {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+
+ // Return null if the value is null
+ if (_segmentColumnReader.isNull(_currentIndex)) {
+ _currentIndex++;
+ return null;
+ }
+ _reuseValue = _segmentColumnReader.getValue(_currentIndex);
+ _currentIndex++;
+
+ // Return null if the value is null
+ if (_reuseValue == null) {
+ return null;
+ }
+
+ return _reuseValue;
+ }
+
+ @Override
+ public boolean isNextNull() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ return _segmentColumnReader.isNull(_currentIndex);
+ }
+
+ @Override
+ public void skipNext() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ _currentIndex++;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return _segmentColumnReader.isSingleValue();
+ }
+
+ @Override
+ public boolean isInt() {
+ return _dataType == FieldSpec.DataType.INT;
+ }
+
+ @Override
+ public boolean isLong() {
+ return _dataType == FieldSpec.DataType.LONG;
+ }
+
+ @Override
+ public boolean isFloat() {
+ return _dataType == FieldSpec.DataType.FLOAT;
+ }
+
+ @Override
+ public boolean isDouble() {
+ return _dataType == FieldSpec.DataType.DOUBLE;
+ }
+
+ @Override
+ public boolean isString() {
+ return _dataType == FieldSpec.DataType.STRING;
+ }
+
+ @Override
+ public boolean isBytes() {
+ return _dataType == FieldSpec.DataType.BYTES;
+ }
+
+ @Override
+ public int nextInt() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ int value = _segmentColumnReader.getInt(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public long nextLong() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ long value = _segmentColumnReader.getLong(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public float nextFloat() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ float value = _segmentColumnReader.getFloat(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public double nextDouble() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ double value = _segmentColumnReader.getDouble(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public String nextString() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ String value = _segmentColumnReader.getString(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public byte[] nextBytes() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ byte[] value = _segmentColumnReader.getBytes(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ // For all multi-value primitive type methods (nextIntMV, nextLongMV, nextFloatMV, nextDoubleMV,
+ // getIntMV, getLongMV, getFloatMV, getDoubleMV), we pass null for the validity bitset since
+ // multi-value primitive types cannot have null elements. Nulls are removed by NullValueTransformer
+ @Override
+ public MultiValueResult nextIntMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ int[] value = _segmentColumnReader.getIntMV(_currentIndex);
+ _currentIndex++;
+ return MultiValueResult.of(value, null);
+ }
+
+ @Override
+ public MultiValueResult nextLongMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ long[] value = _segmentColumnReader.getLongMV(_currentIndex);
+ _currentIndex++;
+ return MultiValueResult.of(value, null);
+ }
+
+ @Override
+ public MultiValueResult nextFloatMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ float[] value = _segmentColumnReader.getFloatMV(_currentIndex);
+ _currentIndex++;
+ return MultiValueResult.of(value, null);
+ }
+
+ @Override
+ public MultiValueResult nextDoubleMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ double[] value = _segmentColumnReader.getDoubleMV(_currentIndex);
+ _currentIndex++;
+ return MultiValueResult.of(value, null);
+ }
+
+ @Override
+ public String[] nextStringMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ String[] value = _segmentColumnReader.getStringMV(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public byte[][] nextBytesMV() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more values available");
+ }
+ byte[][] value = _segmentColumnReader.getBytesMV(_currentIndex);
+ _currentIndex++;
+ return value;
+ }
+
+ @Override
+ public void rewind()
+ throws IOException {
+ _currentIndex = 0;
+ _reuseValue = null;
+ }
+
+ @Override
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ @Override
+ public int getTotalDocs() {
+ return _numDocs;
+ }
+
+ @Override
+ public boolean isNull(int docId) {
+ validateDocId(docId);
+ return _segmentColumnReader.isNull(docId);
+ }
+
+ // Single-value accessors
+
+ @Override
+ public int getInt(int docId) {
+ return _segmentColumnReader.getInt(docId);
+ }
+
+ @Override
+ public long getLong(int docId) {
+ return _segmentColumnReader.getLong(docId);
+ }
+
+ @Override
+ public float getFloat(int docId) {
+ return _segmentColumnReader.getFloat(docId);
+ }
+
+ @Override
+ public double getDouble(int docId) {
+ return _segmentColumnReader.getDouble(docId);
+ }
+
+ @Override
+ public String getString(int docId) {
+ return _segmentColumnReader.getString(docId);
+ }
+
+ @Override
+ public byte[] getBytes(int docId) {
+ return _segmentColumnReader.getBytes(docId);
+ }
+
+ @Override
+ public Object getValue(int docId)
+ throws IOException {
+ return _segmentColumnReader.getValue(docId);
+ }
+
+ // Multi-value accessors
+
+ @Override
+ public MultiValueResult getIntMV(int docId) {
+ return MultiValueResult.of(_segmentColumnReader.getIntMV(docId), null);
+ }
+
+ @Override
+ public MultiValueResult getLongMV(int docId) {
+ return MultiValueResult.of(_segmentColumnReader.getLongMV(docId), null);
+ }
+
+ @Override
+ public MultiValueResult getFloatMV(int docId) {
+ return MultiValueResult.of(_segmentColumnReader.getFloatMV(docId), null);
+ }
+
+ @Override
+ public MultiValueResult getDoubleMV(int docId) {
+ return MultiValueResult.of(_segmentColumnReader.getDoubleMV(docId), null);
+ }
+
+ @Override
+ public String[] getStringMV(int docId) {
+ return _segmentColumnReader.getStringMV(docId);
+ }
+
+ @Override
+ public byte[][] getBytesMV(int docId) {
+ return _segmentColumnReader.getBytesMV(docId);
+ }
+
+ /**
+ * Validate that the document ID is within valid range.
+ *
+ * @param docId Document ID to validate
+ * @throws IndexOutOfBoundsException if docId is out of range
+ */
+ private void validateDocId(int docId) {
+ if (docId < 0 || docId >= _numDocs) {
+ throw new IndexOutOfBoundsException(
+ "docId " + docId + " is out of range. Valid range is 0 to " + (_numDocs - 1));
+ }
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ _segmentColumnReader.close();
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnarDataSource.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnarDataSource.java
new file mode 100644
index 0000000000..960dc7ebc4
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentColumnarDataSource.java
@@ -0,0 +1,63 @@
+/**
+ * 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.segment.local.segment.readers;
+
+import java.io.IOException;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.spi.data.readers.ColumnReaderFactory;
+import org.apache.pinot.spi.data.readers.ColumnarDataSource;
+
+
+/**
+ * ColumnarDataSource implementation that wraps a Pinot segment.
+ * Provides columnar access to segment data via ColumnReaderFactory.
+ */
+public class PinotSegmentColumnarDataSource implements ColumnarDataSource {
+
+ private final IndexSegment _indexSegment;
+ private final int _totalDocs;
+ private final boolean _initializeDefaultValueReaders;
+
+ public PinotSegmentColumnarDataSource(IndexSegment indexSegment, boolean initializeDefaultValueReaders) {
+ _indexSegment = indexSegment;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
+ _initializeDefaultValueReaders = initializeDefaultValueReaders;
+ }
+
+ @Override
+ public int getTotalDocs() {
+ return _totalDocs;
+ }
+
+ @Override
+ public ColumnReaderFactory createColumnReaderFactory() {
+ return new PinotSegmentColumnReaderFactory(_indexSegment, _initializeDefaultValueReaders);
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ // Segment lifecycle is managed externally, so no cleanup needed here
+ }
+
+ @Override
+ public String toString() {
+ return "PinotSegmentColumnarDataSource{segment=" + _indexSegment.getSegmentName() + "}";
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnReaderInterfaceTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnReaderInterfaceTest.java
new file mode 100644
index 0000000000..42f280e170
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnReaderInterfaceTest.java
@@ -0,0 +1,129 @@
+/**
+ * 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.segment.local.segment.creator.impl;
+
+import java.io.File;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReaderFactory;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.spi.data.readers.ColumnReader;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for core ColumnReader interface and implementations.
+ *
+ *
This test class validates:
+ *
+ *
PinotSegmentColumnReaderFactory functionality
+ *
DefaultValueColumnReader for new columns
+ *
ColumnReader interface contract
+ *
+ */
+public class ColumnReaderInterfaceTest extends ColumnarSegmentBuildingTestBase {
+
+ @Test
+ public void testColumnReaderFactory()
+ throws Exception {
+ // Create a segment to test column reader factory with
+ File segmentDir = createRowMajorSegment();
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, ReadMode.mmap);
+
+ try {
+ // Test PinotSegmentColumnReaderFactory
+ try (PinotSegmentColumnReaderFactory factory = new PinotSegmentColumnReaderFactory(segment)) {
+ factory.init(_originalSchema);
+
+ // Test that all expected columns are available
+ Set availableColumns = factory.getAvailableColumns();
+ Assert.assertTrue(availableColumns.contains(STRING_COL_1));
+ Assert.assertTrue(availableColumns.contains(INT_COL_1));
+ Assert.assertTrue(availableColumns.contains(MV_INT_COL));
+
+ // Test getting individual column readers
+ ColumnReader stringReader = factory.getColumnReader(STRING_COL_1);
+ Assert.assertEquals(stringReader.getColumnName(), STRING_COL_1);
+
+ ColumnReader mvReader = factory.getColumnReader(MV_INT_COL);
+ Assert.assertEquals(mvReader.getColumnName(), MV_INT_COL);
+
+ // Test reading values using iterator pattern
+ Assert.assertTrue(stringReader.hasNext());
+ Object firstStringValue = stringReader.next();
+ Assert.assertEquals(firstStringValue, "string1_0");
+
+ // Test that we can continue reading
+ Assert.assertTrue(stringReader.hasNext());
+ Object secondStringValue = stringReader.next();
+ Assert.assertEquals(secondStringValue, "string1_1");
+
+ // Reset and test again
+ stringReader.rewind();
+ Assert.assertTrue(stringReader.hasNext());
+ firstStringValue = stringReader.next();
+ Assert.assertEquals(firstStringValue, "string1_0");
+
+ // Test getting all column readers
+ Map allReaders = factory.getAllColumnReaders();
+ Assert.assertEquals(allReaders.size(), _originalSchema.getPhysicalColumnNames().size());
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ @Test
+ public void testColumnReaderWithNewColumns()
+ throws Exception {
+ // Create a segment to test with
+ File segmentDir = createRowMajorSegment();
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, ReadMode.mmap);
+
+ try {
+ // Test getting readers for new columns (should return default value readers)
+ try (PinotSegmentColumnReaderFactory factory = new PinotSegmentColumnReaderFactory(segment)) {
+ factory.init(_extendedSchema);
+
+ // Test getting reader for new column
+ ColumnReader newStringReader = factory.getColumnReader(NEW_STRING_COL);
+ Assert.assertEquals(newStringReader.getColumnName(), NEW_STRING_COL);
+
+ // Verify it returns default values using iterator pattern
+ Assert.assertTrue(newStringReader.hasNext());
+ Object defaultValue = newStringReader.next();
+ Assert.assertEquals(defaultValue, _extendedSchema.getFieldSpecFor(NEW_STRING_COL).getDefaultNullValue());
+
+ // Test that all values are the same (default)
+ int valueCount = 1; // We already read one value
+ while (newStringReader.hasNext()) {
+ Object value = newStringReader.next();
+ Assert.assertEquals(value, defaultValue);
+ valueCount++;
+ }
+ Assert.assertEquals(valueCount, _testData.size());
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarRowMajorEquivalenceTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarRowMajorEquivalenceTest.java
new file mode 100644
index 0000000000..ac2cac33f7
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarRowMajorEquivalenceTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.segment.local.segment.creator.impl;
+
+import java.io.File;
+import java.util.Set;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for columnar vs row-major equivalence.
+ *
+ *
This test class validates that:
+ *
+ *
Columnar segment building produces identical results to row-major building
+ *
All data types are correctly handled in columnar mode
+ *
Segment statistics and metadata are preserved
+ *
+ */
+public class ColumnarRowMajorEquivalenceTest extends ColumnarSegmentBuildingTestBase {
+
+ @Test
+ public void testBasicColumnarBuilding()
+ throws Exception {
+ // First create a segment using traditional row-major approach
+ File rowMajorSegmentDir = createRowMajorSegment();
+
+ // Then create a segment using columnar approach from the row-major segment
+ File columnarSegmentDir = createColumnarSegment(rowMajorSegmentDir);
+
+ // Validate that both segments have identical data
+ validateSegmentsIdentical(rowMajorSegmentDir, columnarSegmentDir);
+ }
+
+ @Test
+ public void testAllDataTypes()
+ throws Exception {
+ // This test validates that all supported data types work correctly with columnar building
+ File rowMajorSegmentDir = createRowMajorSegment();
+ File columnarSegmentDir = createColumnarSegment(rowMajorSegmentDir);
+
+ // Validate that both segments have identical data for all data types
+ validateSegmentsIdentical(rowMajorSegmentDir, columnarSegmentDir);
+
+ // Additionally validate that all expected columns and data types are present
+ ImmutableSegment segment = ImmutableSegmentLoader.load(columnarSegmentDir, ReadMode.mmap);
+ try {
+ Set columnNames = segment.getPhysicalColumnNames();
+
+ // Validate all data types are present
+ Assert.assertTrue(columnNames.contains(STRING_COL_1), "STRING column missing");
+ Assert.assertTrue(columnNames.contains(INT_COL_1), "INT column missing");
+ Assert.assertTrue(columnNames.contains(LONG_COL), "LONG column missing");
+ Assert.assertTrue(columnNames.contains(FLOAT_COL), "FLOAT column missing");
+ Assert.assertTrue(columnNames.contains(DOUBLE_COL), "DOUBLE column missing");
+ Assert.assertTrue(columnNames.contains(BIG_DECIMAL_COL), "BIG_DECIMAL column missing");
+ Assert.assertTrue(columnNames.contains(BYTES_COL), "BYTES column missing");
+ Assert.assertTrue(columnNames.contains(MV_INT_COL), "Multi-value INT column missing");
+ Assert.assertTrue(columnNames.contains(MV_STRING_COL), "Multi-value STRING column missing");
+ Assert.assertTrue(columnNames.contains(TIME_COL), "TIME column missing");
+ } finally {
+ segment.destroy();
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSchemaEvolutionTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSchemaEvolutionTest.java
new file mode 100644
index 0000000000..ab533bb32c
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSchemaEvolutionTest.java
@@ -0,0 +1,55 @@
+/**
+ * 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.segment.local.segment.creator.impl;
+
+import java.io.File;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for schema evolution and new column handling.
+ *
+ *
This test class validates:
+ *
+ *
Adding new columns with default values during columnar building
+ *
Schema evolution compatibility between row-major and columnar approaches
+ *
Default value handling for single-value and multi-value columns
+ *
+ */
+public class ColumnarSchemaEvolutionTest extends ColumnarSegmentBuildingTestBase {
+
+ @Test
+ public void testColumnarBuildingWithNewColumns()
+ throws Exception {
+ // Create original segment with original schema
+ File originalSegmentDir = createRowMajorSegment();
+
+ // Create columnar segment with extended schema (has additional columns)
+ File columnarSegmentDir = createColumnarSegmentWithNewColumns(originalSegmentDir);
+
+ // Create row-major segment with extended schema for comparison using the same original segment
+ File rowMajorExtendedSegmentDir = createRowMajorSegmentWithExtendedSchema(originalSegmentDir);
+
+ // Validate that both segments are identical (comprehensive comparison)
+ validateSegmentsIdentical(rowMajorExtendedSegmentDir, columnarSegmentDir);
+
+ // Additional validation that the new segment has the additional columns with default values
+ validateSegmentWithNewColumns(columnarSegmentDir);
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSegmentBuildingTestBase.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSegmentBuildingTestBase.java
new file mode 100644
index 0000000000..82dd918124
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarSegmentBuildingTestBase.java
@@ -0,0 +1,578 @@
+/**
+ * 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.segment.local.segment.creator.impl;
+
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReaderFactory;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+
+
+/**
+ * Base class for columnar segment building tests containing common setup and utility methods.
+ *
+ *
This base class provides:
+ *
+ *
Common test data and schema setup
+ *
Segment creation utility methods
+ *
Validation utility methods
+ *
Test data generation
+ *
+ */
+public abstract class ColumnarSegmentBuildingTestBase {
+ protected static final String TEMP_DIR = System.getProperty("java.io.tmpdir");
+ protected static final String TABLE_NAME = "testTable";
+ protected static final String SEGMENT_NAME = "testSegment";
+
+ // Test columns
+ protected static final String STRING_COL_1 = "stringCol1";
+ protected static final String STRING_COL_2 = "stringCol2";
+ protected static final String INT_COL_1 = "intCol1";
+ protected static final String INT_COL_2 = "intCol2";
+ protected static final String LONG_COL = "longCol";
+ protected static final String FLOAT_COL = "floatCol";
+ protected static final String DOUBLE_COL = "doubleCol";
+ protected static final String BIG_DECIMAL_COL = "bigDecimalCol";
+ protected static final String BYTES_COL = "bytesCol";
+ protected static final String TIME_COL = "timeCol";
+ protected static final String MV_INT_COL = "mvIntCol";
+ protected static final String MV_LONG_COL = "mvLongCol";
+ protected static final String MV_FLOAT_COL = "mvFloatCol";
+ protected static final String MV_DOUBLE_COL = "mvDoubleCol";
+ protected static final String MV_STRING_COL = "mvStringCol";
+ protected static final String MV_BYTES_COL = "mvBytesCol";
+
+ // New column for testing default value handling
+ protected static final String NEW_STRING_COL = "newStringCol";
+ protected static final String NEW_INT_COL = "newIntCol";
+ protected static final String NEW_MV_LONG_COL = "newMvLongCol";
+
+ protected File _tempDir;
+ protected Schema _originalSchema;
+ protected Schema _extendedSchema; // Schema with additional columns
+ protected TableConfig _tableConfig;
+ protected List _testData;
+
+ @BeforeClass
+ public void setUp()
+ throws IOException {
+ _tempDir = new File(TEMP_DIR, getClass().getSimpleName());
+ FileUtils.deleteQuietly(_tempDir);
+ _tempDir.mkdirs();
+
+ // Create original schema
+ _originalSchema = new Schema.SchemaBuilder()
+ .addSingleValueDimension(STRING_COL_1, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(STRING_COL_2, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(INT_COL_1, FieldSpec.DataType.INT)
+ .addSingleValueDimension(INT_COL_2, FieldSpec.DataType.INT)
+ .addSingleValueDimension(LONG_COL, FieldSpec.DataType.LONG)
+ .addSingleValueDimension(FLOAT_COL, FieldSpec.DataType.FLOAT)
+ .addSingleValueDimension(DOUBLE_COL, FieldSpec.DataType.DOUBLE)
+ .addSingleValueDimension(BIG_DECIMAL_COL, FieldSpec.DataType.BIG_DECIMAL)
+ .addSingleValueDimension(BYTES_COL, FieldSpec.DataType.BYTES)
+ .addMultiValueDimension(MV_INT_COL, FieldSpec.DataType.INT)
+ .addMultiValueDimension(MV_LONG_COL, FieldSpec.DataType.LONG)
+ .addMultiValueDimension(MV_FLOAT_COL, FieldSpec.DataType.FLOAT)
+ .addMultiValueDimension(MV_DOUBLE_COL, FieldSpec.DataType.DOUBLE)
+ .addMultiValueDimension(MV_STRING_COL, FieldSpec.DataType.STRING)
+ .addMultiValueDimension(MV_BYTES_COL, FieldSpec.DataType.BYTES)
+ .addDateTime(TIME_COL, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .build();
+
+ // Create extended schema with additional columns
+ _extendedSchema = new Schema.SchemaBuilder()
+ .addSingleValueDimension(STRING_COL_1, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(STRING_COL_2, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(INT_COL_1, FieldSpec.DataType.INT)
+ .addSingleValueDimension(INT_COL_2, FieldSpec.DataType.INT)
+ .addSingleValueDimension(LONG_COL, FieldSpec.DataType.LONG)
+ .addSingleValueDimension(FLOAT_COL, FieldSpec.DataType.FLOAT)
+ .addSingleValueDimension(DOUBLE_COL, FieldSpec.DataType.DOUBLE)
+ .addSingleValueDimension(BIG_DECIMAL_COL, FieldSpec.DataType.BIG_DECIMAL)
+ .addSingleValueDimension(BYTES_COL, FieldSpec.DataType.BYTES)
+ .addMultiValueDimension(MV_INT_COL, FieldSpec.DataType.INT)
+ .addMultiValueDimension(MV_LONG_COL, FieldSpec.DataType.LONG)
+ .addMultiValueDimension(MV_FLOAT_COL, FieldSpec.DataType.FLOAT)
+ .addMultiValueDimension(MV_DOUBLE_COL, FieldSpec.DataType.DOUBLE)
+ .addMultiValueDimension(MV_STRING_COL, FieldSpec.DataType.STRING)
+ .addMultiValueDimension(MV_BYTES_COL, FieldSpec.DataType.BYTES)
+ .addDateTime(TIME_COL, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .addSingleValueDimension(NEW_STRING_COL, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(NEW_INT_COL, FieldSpec.DataType.INT)
+ .addMultiValueDimension(NEW_MV_LONG_COL, FieldSpec.DataType.LONG)
+ .build();
+
+ // Create table config
+ _tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COL)
+ .setInvertedIndexColumns(Lists.newArrayList(STRING_COL_1, INT_COL_1))
+ .setSortedColumn(INT_COL_1)
+ .build();
+
+ // Generate test data
+ _testData = generateTestData(100);
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws IOException {
+ FileUtils.deleteQuietly(_tempDir);
+ }
+
+ protected File createRowMajorSegment()
+ throws Exception {
+ return createRowMajorSegmentWithConfig(_tableConfig, _originalSchema, "rowMajorSegment");
+ }
+
+ protected File createRowMajorSegmentWithConfig(TableConfig tableConfig, Schema schema, String dirName)
+ throws Exception {
+ File outputDir = new File(_tempDir, dirName);
+ FileUtils.deleteQuietly(outputDir);
+ outputDir.mkdirs();
+
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, schema);
+ config.setOutDir(outputDir.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME + "_" + dirName);
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ TestRecordReader recordReader = new TestRecordReader(_testData);
+ driver.init(config, recordReader);
+ driver.build();
+
+ return new File(outputDir, SEGMENT_NAME + "_" + dirName);
+ }
+
+ protected TableConfig createTableConfigWithNoDictionary() {
+ List allColumns = new ArrayList<>();
+ allColumns.add(STRING_COL_1);
+ allColumns.add(STRING_COL_2);
+ allColumns.add(INT_COL_1);
+ allColumns.add(INT_COL_2);
+ allColumns.add(LONG_COL);
+ allColumns.add(FLOAT_COL);
+ allColumns.add(DOUBLE_COL);
+ allColumns.add(BIG_DECIMAL_COL);
+ allColumns.add(BYTES_COL);
+ allColumns.add(MV_INT_COL);
+ allColumns.add(MV_LONG_COL);
+ allColumns.add(MV_FLOAT_COL);
+ allColumns.add(MV_DOUBLE_COL);
+ allColumns.add(MV_STRING_COL);
+ allColumns.add(MV_BYTES_COL);
+
+ return new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(TABLE_NAME)
+ .setTimeColumnName(TIME_COL)
+ .setNoDictionaryColumns(allColumns)
+ .build();
+ }
+
+ protected Schema createNullableSchema() {
+ return new Schema.SchemaBuilder()
+ .setEnableColumnBasedNullHandling(true)
+ .addDimensionField(STRING_COL_1, FieldSpec.DataType.STRING, spec -> spec.setNullable(true))
+ .addDimensionField(STRING_COL_2, FieldSpec.DataType.STRING, spec -> spec.setNullable(true))
+ .addDimensionField(INT_COL_1, FieldSpec.DataType.INT, spec -> spec.setNullable(true))
+ .addDimensionField(INT_COL_2, FieldSpec.DataType.INT, spec -> spec.setNullable(true))
+ .addDimensionField(LONG_COL, FieldSpec.DataType.LONG, spec -> spec.setNullable(true))
+ .addDimensionField(FLOAT_COL, FieldSpec.DataType.FLOAT, spec -> spec.setNullable(true))
+ .addDimensionField(DOUBLE_COL, FieldSpec.DataType.DOUBLE, spec -> spec.setNullable(true))
+ .addDimensionField(BIG_DECIMAL_COL, FieldSpec.DataType.BIG_DECIMAL, spec -> spec.setNullable(true))
+ .addDimensionField(BYTES_COL, FieldSpec.DataType.BYTES, spec -> spec.setNullable(true))
+ .addDimensionField(MV_INT_COL, FieldSpec.DataType.INT, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDimensionField(MV_LONG_COL, FieldSpec.DataType.LONG, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDimensionField(MV_FLOAT_COL, FieldSpec.DataType.FLOAT, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDimensionField(MV_DOUBLE_COL, FieldSpec.DataType.DOUBLE, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDimensionField(MV_STRING_COL, FieldSpec.DataType.STRING, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDimensionField(MV_BYTES_COL, FieldSpec.DataType.BYTES, spec -> {
+ spec.setSingleValueField(false);
+ spec.setNullable(true);
+ })
+ .addDateTime(TIME_COL, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ .build();
+ }
+
+ protected File createColumnarSegment(File sourceSegmentDir)
+ throws Exception {
+ File outputDir = new File(_tempDir, "columnarSegment");
+ FileUtils.deleteQuietly(outputDir);
+ outputDir.mkdirs();
+
+ // Load the source segment
+ ImmutableSegment sourceSegment = ImmutableSegmentLoader.load(sourceSegmentDir, ReadMode.mmap);
+
+ try {
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, _originalSchema);
+ config.setOutDir(outputDir.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME + "_columnar");
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+
+ // Use the new columnar building approach with config
+ try (PinotSegmentColumnReaderFactory factory = new PinotSegmentColumnReaderFactory(sourceSegment)) {
+ driver.init(config, factory);
+ driver.build();
+ }
+
+ return new File(outputDir, SEGMENT_NAME + "_columnar");
+ } finally {
+ sourceSegment.destroy();
+ }
+ }
+
+ protected File createColumnarSegmentWithNewColumns(File sourceSegmentDir)
+ throws Exception {
+ File outputDir = new File(_tempDir, "columnarSegmentWithNewColumns");
+ FileUtils.deleteQuietly(outputDir);
+ outputDir.mkdirs();
+
+ // Load the source segment
+ ImmutableSegment sourceSegment = ImmutableSegmentLoader.load(sourceSegmentDir, ReadMode.mmap);
+
+ try {
+ // Use extended schema with new columns
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, _extendedSchema);
+ config.setOutDir(outputDir.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME + "_withNewColumns");
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+
+ // Use columnar building with extended schema
+ try (PinotSegmentColumnReaderFactory factory = new PinotSegmentColumnReaderFactory(sourceSegment)) {
+ driver.init(config, factory);
+ driver.build();
+ }
+
+ return new File(outputDir, SEGMENT_NAME + "_withNewColumns");
+ } finally {
+ sourceSegment.destroy();
+ }
+ }
+
+ protected File createRowMajorSegmentWithExtendedSchema(File originalSegmentDir)
+ throws Exception {
+ File outputDir = new File(_tempDir, "rowMajorSegmentWithExtendedSchema");
+ FileUtils.deleteQuietly(outputDir);
+ outputDir.mkdirs();
+
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, _extendedSchema);
+ config.setOutDir(outputDir.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME + "_rowMajorExtended");
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+
+ // Use PinotSegmentRecordReader to read from the original segment, similar to RefreshSegmentTaskExecutor
+ try (PinotSegmentRecordReader recordReader = new PinotSegmentRecordReader()) {
+ recordReader.init(originalSegmentDir, null, null);
+ driver.init(config, recordReader);
+ driver.build();
+ }
+
+ return new File(outputDir, SEGMENT_NAME + "_rowMajorExtended");
+ }
+
+ protected void validateSegmentsIdentical(File segment1Dir, File segment2Dir)
+ throws Exception {
+ ImmutableSegment segment1 = ImmutableSegmentLoader.load(segment1Dir, ReadMode.mmap);
+ ImmutableSegment segment2 = ImmutableSegmentLoader.load(segment2Dir, ReadMode.mmap);
+
+ try {
+ // Validate basic segment metadata
+ Assert.assertEquals(segment1.getSegmentMetadata().getTotalDocs(), segment2.getSegmentMetadata().getTotalDocs());
+ Assert.assertEquals(segment1.getSegmentMetadata().getAllColumns(), segment2.getSegmentMetadata().getAllColumns());
+
+ // Validate column metadata and statistics for each column
+ for (String columnName : segment1.getPhysicalColumnNames()) {
+ validateColumnMetadata(segment1, segment2, columnName);
+ validateColumnData(segment1, segment2, columnName);
+ }
+ } finally {
+ segment1.destroy();
+ segment2.destroy();
+ }
+ }
+
+ protected void validateColumnMetadata(ImmutableSegment segment1, ImmutableSegment segment2, String columnName) {
+ ColumnMetadata metadata1 = segment1.getSegmentMetadata().getColumnMetadataFor(columnName);
+ ColumnMetadata metadata2 = segment2.getSegmentMetadata().getColumnMetadataFor(columnName);
+
+ Assert.assertNotNull(metadata1, "Column metadata missing for " + columnName + " in segment1");
+ Assert.assertNotNull(metadata2, "Column metadata missing for " + columnName + " in segment2");
+
+ // Validate basic column properties
+ Assert.assertEquals(metadata1.getDataType(), metadata2.getDataType(),
+ "Data type mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.isSingleValue(), metadata2.isSingleValue(),
+ "Single value flag mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.getFieldType(), metadata2.getFieldType(),
+ "Field type mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.getTotalDocs(), metadata2.getTotalDocs(),
+ "Total docs mismatch for column " + columnName);
+
+ // Validate cardinality and dictionary properties
+ Assert.assertEquals(metadata1.getCardinality(), metadata2.getCardinality(),
+ "Cardinality mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.hasDictionary(), metadata2.hasDictionary(),
+ "Dictionary flag mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.getColumnMaxLength(), metadata2.getColumnMaxLength(),
+ "Column max length mismatch for column " + columnName);
+
+ // Validate sorting and indexing properties
+ Assert.assertEquals(metadata1.isSorted(), metadata2.isSorted(),
+ "Sorted flag mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.isAutoGenerated(), metadata2.isAutoGenerated(),
+ "Auto-generated flag mismatch for column " + columnName);
+
+ // Validate multi-value properties
+ if (!metadata1.isSingleValue()) {
+ Assert.assertEquals(metadata1.getMaxNumberOfMultiValues(), metadata2.getMaxNumberOfMultiValues(),
+ "Max number of multi-values mismatch for column " + columnName);
+ Assert.assertEquals(metadata1.getTotalNumberOfEntries(), metadata2.getTotalNumberOfEntries(),
+ "Total number of entries mismatch for column " + columnName);
+ }
+
+ // Validate min/max values if available
+ Comparable> minValue1 = metadata1.getMinValue();
+ Comparable> minValue2 = metadata2.getMinValue();
+ Comparable> maxValue1 = metadata1.getMaxValue();
+ Comparable> maxValue2 = metadata2.getMaxValue();
+
+ if (minValue1 != null && minValue2 != null) {
+ Assert.assertEquals(minValue1, minValue2,
+ "Min value mismatch for column " + columnName);
+ }
+ if (maxValue1 != null && maxValue2 != null) {
+ Assert.assertEquals(maxValue1, maxValue2,
+ "Max value mismatch for column " + columnName);
+ }
+
+ // Validate bits per element for dictionary encoded columns
+ if (metadata1.hasDictionary() && metadata2.hasDictionary()) {
+ Assert.assertEquals(metadata1.getBitsPerElement(), metadata2.getBitsPerElement(),
+ "Bits per element mismatch for column " + columnName);
+ }
+ }
+
+ protected void validateSegmentWithNewColumns(File segmentDir)
+ throws Exception {
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, ReadMode.mmap);
+
+ try {
+ // Validate that new columns exist
+ Assert.assertTrue(segment.getPhysicalColumnNames().contains(NEW_STRING_COL));
+ Assert.assertTrue(segment.getPhysicalColumnNames().contains(NEW_INT_COL));
+ Assert.assertTrue(segment.getPhysicalColumnNames().contains(NEW_MV_LONG_COL));
+
+ // Validate that new columns have default values
+ GenericRow row = new GenericRow();
+ for (int docId = 0; docId < segment.getSegmentMetadata().getTotalDocs(); docId++) {
+ segment.getRecord(docId, row);
+
+ // Check default values
+ Assert.assertEquals(row.getValue(NEW_STRING_COL),
+ _extendedSchema.getFieldSpecFor(NEW_STRING_COL).getDefaultNullValue());
+ Assert.assertEquals(row.getValue(NEW_INT_COL),
+ _extendedSchema.getFieldSpecFor(NEW_INT_COL).getDefaultNullValue());
+
+ // For multi-value columns, the default value is wrapped in an array
+ Object mvLongValue = row.getValue(NEW_MV_LONG_COL);
+ Assert.assertTrue(mvLongValue instanceof Object[], "Multi-value column should return an array");
+ Object[] mvLongArray = (Object[]) mvLongValue;
+ Assert.assertEquals(mvLongArray.length, 1, "Default multi-value array should have one element");
+ Assert.assertEquals(mvLongArray[0],
+ _extendedSchema.getFieldSpecFor(NEW_MV_LONG_COL).getDefaultNullValue());
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ protected void validateColumnData(ImmutableSegment segment1, ImmutableSegment segment2, String columnName) {
+ int numDocs = segment1.getSegmentMetadata().getTotalDocs();
+
+ GenericRow row1 = new GenericRow();
+ GenericRow row2 = new GenericRow();
+
+ for (int docId = 0; docId < numDocs; docId++) {
+ segment1.getRecord(docId, row1);
+ segment2.getRecord(docId, row2);
+
+ Object value1 = row1.getValue(columnName);
+ Object value2 = row2.getValue(columnName);
+
+ Assert.assertEquals(value1, value2,
+ String.format("Column %s differs at docId %d: %s vs %s", columnName, docId, value1, value2));
+ }
+ }
+
+ protected List generateTestData(int numRows) {
+ List data = new ArrayList<>();
+
+ // Use a fixed seed for reproducible test results
+ Random random = new Random(42);
+
+ // Null probability - approximately 10% of values will be null for nullable columns
+ double nullProbability = 0.1;
+
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = new GenericRow();
+
+ // STRING columns - can be null
+ row.putValue(STRING_COL_1, random.nextDouble() < nullProbability ? null : "string1_" + i);
+ row.putValue(STRING_COL_2, random.nextDouble() < nullProbability ? null : "string2_" + (i % 10));
+
+ // INT columns - can be null
+ row.putValue(INT_COL_1, random.nextDouble() < nullProbability ? null : i);
+ row.putValue(INT_COL_2, random.nextDouble() < nullProbability ? null : i * 2);
+
+ // LONG column - can be null
+ row.putValue(LONG_COL, random.nextDouble() < nullProbability ? null : (long) i * 3);
+
+ // FLOAT column - can be null
+ row.putValue(FLOAT_COL, random.nextDouble() < nullProbability ? null : (float) i * 1.5);
+
+ // DOUBLE column - can be null
+ row.putValue(DOUBLE_COL, random.nextDouble() < nullProbability ? null : (double) i * 2.5);
+
+ // BIG_DECIMAL column - can be null
+ row.putValue(BIG_DECIMAL_COL, random.nextDouble() < nullProbability ? null : new BigDecimal(i + ".123"));
+
+ // BYTES column - can be null
+ row.putValue(BYTES_COL, random.nextDouble() < nullProbability ? null : ("bytes_" + i).getBytes());
+
+ // TIME column - typically not null for time columns, but we can test it
+ // Use a deterministic timestamp based on the row index for consistent results
+ long baseTimestamp = 1700000000000L; // Fixed base timestamp
+ row.putValue(TIME_COL, random.nextDouble() < nullProbability ? null : baseTimestamp + i);
+
+ // Multi-value columns - can be null (entire array is null, not individual elements)
+ row.putValue(MV_INT_COL, random.nextDouble() < nullProbability ? null : new Object[]{i, i + 1, i + 2});
+ row.putValue(MV_LONG_COL, random.nextDouble() < nullProbability ? null
+ : new Object[]{(long) i * 10, (long) (i + 1) * 10, (long) (i + 2) * 10});
+ row.putValue(MV_FLOAT_COL, random.nextDouble() < nullProbability ? null
+ : new Object[]{(float) i * 1.1f, (float) (i + 1) * 1.1f, (float) (i + 2) * 1.1f});
+ row.putValue(MV_DOUBLE_COL, random.nextDouble() < nullProbability ? null
+ : new Object[]{(double) i * 2.2, (double) (i + 1) * 2.2, (double) (i + 2) * 2.2});
+ row.putValue(MV_STRING_COL, random.nextDouble() < nullProbability ? null
+ : new Object[]{"mv1_" + i, "mv2_" + i, "mv3_" + (i % 3)});
+ row.putValue(MV_BYTES_COL, random.nextDouble() < nullProbability ? null
+ : new Object[]{("mvbytes1_" + i).getBytes(), ("mvbytes2_" + i).getBytes()});
+
+ data.add(row);
+ }
+
+ return data;
+ }
+
+ /**
+ * Simple test record reader for the test data.
+ */
+ protected static class TestRecordReader implements RecordReader {
+ private final List _data;
+ private int _currentIndex = 0;
+
+ public TestRecordReader(List data) {
+ _data = data;
+ }
+
+ @Override
+ public void init(File dataFile, @Nullable Set fieldsToRead,
+ @Nullable RecordReaderConfig recordReaderConfig) {
+ _currentIndex = 0;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return _currentIndex < _data.size();
+ }
+
+ @Override
+ public GenericRow next() {
+ return next(new GenericRow());
+ }
+
+ @Override
+ public GenericRow next(GenericRow reuse) {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more records");
+ }
+
+ GenericRow sourceRow = _data.get(_currentIndex++);
+ reuse.clear();
+
+ for (Map.Entry entry : sourceRow.getFieldToValueMap().entrySet()) {
+ reuse.putValue(entry.getKey(), entry.getValue());
+ }
+
+ return reuse;
+ }
+
+ @Override
+ public void rewind() {
+ _currentIndex = 0;
+ }
+
+ @Override
+ public void close() {
+ // No-op
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollectorTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollectorTest.java
new file mode 100644
index 0000000000..3be1c48aaf
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/AbstractColumnStatisticsCollectorTest.java
@@ -0,0 +1,712 @@
+/**
+ * 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.segment.local.segment.creator.impl.stats;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.Set;
+import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Comprehensive tests for AbstractColumnStatisticsCollector implementations.
+ * Tests all data types with both primitive and Object collect methods,
+ * single and multi-value scenarios, edge cases, and partition functionality.
+ */
+public class AbstractColumnStatisticsCollectorTest {
+
+ private static final String COLUMN_NAME = "testColumn";
+ private static final String TABLE_NAME = "testTable";
+ private static final int NUM_PARTITIONS = 4;
+
+ // Test helper methods
+
+ /**
+ * Creates a StatsCollectorConfig with optional partitioning.
+ */
+ private StatsCollectorConfig createStatsCollectorConfig(FieldSpec.DataType dataType, boolean isSingleValue,
+ boolean withPartitioning) {
+ TableConfigBuilder builder = new TableConfigBuilder(org.apache.pinot.spi.config.table.TableType.OFFLINE)
+ .setTableName(TABLE_NAME);
+
+ if (withPartitioning) {
+ builder.setSegmentPartitionConfig(new SegmentPartitionConfig(
+ Collections.singletonMap(COLUMN_NAME, new ColumnPartitionConfig("Murmur", NUM_PARTITIONS))));
+ }
+
+ TableConfig tableConfig = builder.build();
+ Schema schema = new Schema();
+ schema.addField(new DimensionFieldSpec(COLUMN_NAME, dataType, isSingleValue));
+
+ return new StatsCollectorConfig(tableConfig, schema,
+ withPartitioning ? tableConfig.getIndexingConfig().getSegmentPartitionConfig() : null);
+ }
+
+ /**
+ * Factory method to create the appropriate collector for a given data type.
+ */
+ private AbstractColumnStatisticsCollector createCollector(FieldSpec.DataType dataType, boolean isSingleValue,
+ boolean withPartitioning) {
+ StatsCollectorConfig config = createStatsCollectorConfig(dataType, isSingleValue, withPartitioning);
+ switch (dataType) {
+ case INT:
+ return new IntColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ case LONG:
+ return new LongColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ case FLOAT:
+ return new FloatColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ case DOUBLE:
+ return new DoubleColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ case STRING:
+ return new StringColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ case BYTES:
+ return new BytesColumnPredIndexStatsCollector(COLUMN_NAME, config);
+ case BIG_DECIMAL:
+ return new BigDecimalColumnPreIndexStatsCollector(COLUMN_NAME, config);
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ /**
+ * Helper method to assert basic statistics after collection.
+ */
+ private void assertBasicStats(AbstractColumnStatisticsCollector collector, int expectedCardinality,
+ int expectedTotalEntries, Object expectedMin, Object expectedMax, boolean expectedSorted) {
+ assertEquals(collector.getCardinality(), expectedCardinality);
+ assertEquals(collector.getTotalNumberOfEntries(), expectedTotalEntries);
+ assertEquals(collector.isSorted(), expectedSorted);
+
+ if (expectedMin != null && expectedMax != null) {
+ assertEquals(collector.getMinValue(), expectedMin);
+ assertEquals(collector.getMaxValue(), expectedMax);
+ }
+ }
+
+ // DataProviders
+
+ @DataProvider(name = "allDataTypes")
+ public Object[][] allDataTypes() {
+ return new Object[][] {
+ {FieldSpec.DataType.INT},
+ {FieldSpec.DataType.LONG},
+ {FieldSpec.DataType.FLOAT},
+ {FieldSpec.DataType.DOUBLE},
+ {FieldSpec.DataType.STRING},
+ {FieldSpec.DataType.BYTES},
+ {FieldSpec.DataType.BIG_DECIMAL}
+ };
+ }
+
+ @DataProvider(name = "primitiveDataTypes")
+ public Object[][] primitiveDataTypes() {
+ return new Object[][] {
+ {FieldSpec.DataType.INT},
+ {FieldSpec.DataType.LONG},
+ {FieldSpec.DataType.FLOAT},
+ {FieldSpec.DataType.DOUBLE}
+ };
+ }
+
+ @DataProvider(name = "numericDataTypes")
+ public Object[][] numericDataTypes() {
+ return new Object[][] {
+ {FieldSpec.DataType.INT},
+ {FieldSpec.DataType.LONG},
+ {FieldSpec.DataType.FLOAT},
+ {FieldSpec.DataType.DOUBLE},
+ {FieldSpec.DataType.BIG_DECIMAL}
+ };
+ }
+
+ @DataProvider(name = "multiValueSupportedTypes")
+ public Object[][] multiValueSupportedTypes() {
+ return new Object[][] {
+ {FieldSpec.DataType.INT},
+ {FieldSpec.DataType.LONG},
+ {FieldSpec.DataType.FLOAT},
+ {FieldSpec.DataType.DOUBLE},
+ {FieldSpec.DataType.STRING},
+ {FieldSpec.DataType.BYTES}
+ };
+ }
+
+ @DataProvider(name = "lengthTrackingTypes")
+ public Object[][] lengthTrackingTypes() {
+ return new Object[][] {
+ {FieldSpec.DataType.STRING},
+ {FieldSpec.DataType.BYTES},
+ {FieldSpec.DataType.BIG_DECIMAL}
+ };
+ }
+
+ // Test 1: Single-Value Collection using Primitive Methods
+
+ @Test(dataProvider = "primitiveDataTypes")
+ public void testSingleValuePrimitiveCollectionSorted(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect sorted values
+ switch (dataType) {
+ case INT:
+ collector.collect(1);
+ collector.collect(5);
+ collector.collect(5); // duplicate
+ collector.collect(10);
+ break;
+ case LONG:
+ collector.collect(1L);
+ collector.collect(5L);
+ collector.collect(5L); // duplicate
+ collector.collect(10L);
+ break;
+ case FLOAT:
+ collector.collect(1.5f);
+ collector.collect(5.5f);
+ collector.collect(5.5f); // duplicate
+ collector.collect(10.5f);
+ break;
+ case DOUBLE:
+ collector.collect(1.5);
+ collector.collect(5.5);
+ collector.collect(5.5); // duplicate
+ collector.collect(10.5);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+
+ collector.seal();
+
+ // Verify stats
+ assertBasicStats(collector, 3, 4, getMinValue(dataType), getMaxValue(dataType), true);
+
+ // Verify unique values set
+ Object uniqueValuesSet = collector.getUniqueValuesSet();
+ assertNotNull(uniqueValuesSet);
+ assertEquals(getArrayLength(uniqueValuesSet), 3);
+ }
+
+ @Test(dataProvider = "primitiveDataTypes")
+ public void testSingleValuePrimitiveCollectionUnsorted(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect unsorted values
+ switch (dataType) {
+ case INT:
+ collector.collect(10);
+ collector.collect(1);
+ collector.collect(5);
+ collector.collect(5); // duplicate
+ break;
+ case LONG:
+ collector.collect(10L);
+ collector.collect(1L);
+ collector.collect(5L);
+ collector.collect(5L); // duplicate
+ break;
+ case FLOAT:
+ collector.collect(10.5f);
+ collector.collect(1.5f);
+ collector.collect(5.5f);
+ collector.collect(5.5f); // duplicate
+ break;
+ case DOUBLE:
+ collector.collect(10.5);
+ collector.collect(1.5);
+ collector.collect(5.5);
+ collector.collect(5.5); // duplicate
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+
+ collector.seal();
+
+ // Verify stats - should not be sorted
+ assertBasicStats(collector, 3, 4, getMinValue(dataType), getMaxValue(dataType), false);
+ }
+
+ // Test 2: Single-Value Collection using collect(Object)
+
+ @Test(dataProvider = "allDataTypes")
+ public void testSingleValueObjectCollection(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect values using Object method
+ Object[] testData = getTestDataForType(dataType, true);
+ for (Object value : testData) {
+ collector.collect(value);
+ }
+
+ collector.seal();
+
+ // Verify stats
+ int expectedCardinality = 3; // test data has 1 duplicate
+ int expectedTotalEntries = 4;
+ assertBasicStats(collector, expectedCardinality, expectedTotalEntries,
+ getExpectedMinForType(dataType), getExpectedMaxForType(dataType), true);
+ }
+
+ // Test 3: Multi-Value Collection using Primitive Arrays
+
+ @Test(dataProvider = "multiValueSupportedTypes")
+ public void testMultiValuePrimitiveArrayCollection(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, false, false);
+
+ // Collect multi-value entries
+ switch (dataType) {
+ case INT:
+ collector.collect(new int[]{1, 2});
+ collector.collect(new int[]{3, 4, 5});
+ break;
+ case LONG:
+ collector.collect(new long[]{1L, 2L});
+ collector.collect(new long[]{3L, 4L, 5L});
+ break;
+ case FLOAT:
+ collector.collect(new float[]{1.5f, 2.5f});
+ collector.collect(new float[]{3.5f, 4.5f, 5.5f});
+ break;
+ case DOUBLE:
+ collector.collect(new double[]{1.5, 2.5});
+ collector.collect(new double[]{3.5, 4.5, 5.5});
+ break;
+ case STRING:
+ collector.collect(new String[]{"a", "b"});
+ collector.collect(new String[]{"c", "d", "e"});
+ break;
+ case BYTES:
+ collector.collect(new byte[][]{{1}, {2}});
+ collector.collect(new byte[][]{{3}, {4}, {5}});
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+
+ collector.seal();
+
+ // Verify stats
+ assertEquals(collector.getCardinality(), 5);
+ assertEquals(collector.getTotalNumberOfEntries(), 5);
+ assertEquals(collector.getMaxNumberOfMultiValues(), 3);
+ assertFalse(collector.isSorted()); // multi-value is never sorted
+ }
+
+ @Test(dataProvider = "multiValueSupportedTypes")
+ public void testMultiValueObjectArrayCollection(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, false, false);
+
+ // Collect multi-value entries using Object[]
+ Object[][] testData = getMultiValueTestData(dataType);
+ for (Object[] values : testData) {
+ collector.collect((Object) values);
+ }
+
+ collector.seal();
+
+ // Verify stats
+ assertEquals(collector.getCardinality(), 5);
+ assertEquals(collector.getTotalNumberOfEntries(), 5);
+ assertEquals(collector.getMaxNumberOfMultiValues(), 3);
+ assertFalse(collector.isSorted());
+ }
+
+ // Test 4: Edge Cases
+
+ @Test(dataProvider = "allDataTypes")
+ public void testEmptyCollection(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Seal without collecting any data
+ collector.seal();
+
+ // Verify stats
+ assertEquals(collector.getCardinality(), 0);
+ assertEquals(collector.getTotalNumberOfEntries(), 0);
+ assertTrue(collector.isSorted()); // Empty is considered sorted
+ }
+
+ @Test(dataProvider = "allDataTypes")
+ public void testSingleValue(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect single value
+ Object singleValue = getSingleValueForType(dataType);
+ collector.collect(singleValue);
+ collector.seal();
+
+ // Verify stats - for BYTES, need to use ByteArray for comparison
+ Object expectedValue = dataType == FieldSpec.DataType.BYTES
+ ? new ByteArray((byte[]) singleValue)
+ : singleValue;
+ assertBasicStats(collector, 1, 1, expectedValue, expectedValue, true);
+ }
+
+ @Test(dataProvider = "allDataTypes")
+ public void testAllDuplicates(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect same value multiple times
+ Object value = getSingleValueForType(dataType);
+ for (int i = 0; i < 10; i++) {
+ collector.collect(value);
+ }
+ collector.seal();
+
+ // Verify stats - for BYTES, need to use ByteArray for comparison
+ Object expectedValue = dataType == FieldSpec.DataType.BYTES
+ ? new ByteArray((byte[]) value)
+ : value;
+ assertBasicStats(collector, 1, 10, expectedValue, expectedValue, true);
+ }
+
+ @Test(dataProvider = "allDataTypes", expectedExceptions = IllegalStateException.class)
+ public void testUnsealedAccessThrowsException(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Try to access stats before sealing - should throw
+ collector.getMinValue();
+ }
+
+ // Test 5: Partition Function Tests
+
+ @Test(dataProvider = "allDataTypes")
+ public void testPartitionFunctionEnabled(FieldSpec.DataType dataType) {
+ if (dataType == FieldSpec.DataType.BIG_DECIMAL) {
+ // Skip BigDecimal as partition testing is less relevant
+ return;
+ }
+
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, true);
+
+ // Collect values
+ Object[] testData = getTestDataForType(dataType, true);
+ for (Object value : testData) {
+ collector.collect(value);
+ }
+ collector.seal();
+
+ // Verify partition function is present
+ assertNotNull(collector.getPartitionFunction());
+ assertEquals(collector.getNumPartitions(), NUM_PARTITIONS);
+
+ // Verify partitions are populated
+ Set partitions = collector.getPartitions();
+ assertNotNull(partitions);
+ assertTrue(partitions.size() > 0);
+ assertTrue(partitions.size() <= NUM_PARTITIONS);
+
+ // All partition values should be within range
+ for (Integer partition : partitions) {
+ assertTrue(partition >= 0 && partition < NUM_PARTITIONS);
+ }
+ }
+
+ @Test(dataProvider = "allDataTypes")
+ public void testPartitionFunctionDisabled(FieldSpec.DataType dataType) {
+ AbstractColumnStatisticsCollector collector = createCollector(dataType, true, false);
+
+ // Collect values
+ Object[] testData = getTestDataForType(dataType, true);
+ for (Object value : testData) {
+ collector.collect(value);
+ }
+ collector.seal();
+
+ // Verify partition function is not present
+ assertNull(collector.getPartitionFunction());
+ assertNull(collector.getPartitions());
+ assertEquals(collector.getNumPartitions(), -1);
+ }
+
+ // Test 6: Type-Specific Tests
+
+ @Test
+ public void testStringLengthTracking() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.STRING, true, false);
+
+ collector.collect("a"); // length 1
+ collector.collect("xyz"); // length 3
+ collector.collect("hello"); // length 5
+
+ collector.seal();
+
+ assertEquals(collector.getLengthOfLargestElement(), 5);
+ assertEquals(collector.getMaxRowLengthInBytes(), 5);
+ }
+
+ @Test
+ public void testStringMultiValueLengthTracking() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.STRING, false, false);
+
+ collector.collect(new String[]{"a", "xy"}); // row length: 1 + 2 = 3
+ collector.collect(new String[]{"hello", "world"}); // row length: 5 + 5 = 10
+
+ collector.seal();
+
+ assertEquals(collector.getLengthOfLargestElement(), 5);
+ assertEquals(collector.getMaxRowLengthInBytes(), 10);
+ }
+
+ @Test
+ public void testBytesLengthTracking() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.BYTES, true, false);
+
+ collector.collect(new byte[]{1}); // length 1
+ collector.collect(new byte[]{1, 2, 3}); // length 3
+ collector.collect(new byte[]{1, 2, 3, 4, 5}); // length 5
+
+ collector.seal();
+
+ assertEquals(collector.getLengthOfShortestElement(), 1);
+ assertEquals(collector.getLengthOfLargestElement(), 5);
+ assertEquals(collector.getMaxRowLengthInBytes(), 5);
+ }
+
+ @Test
+ public void testBytesMultiValueLengthTracking() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.BYTES, false, false);
+
+ collector.collect(new byte[][]{{1}, {1, 2}}); // row length: 1 + 2 = 3
+ collector.collect(new byte[][]{{1, 2, 3, 4, 5}, {1, 2, 3, 4, 5}}); // row length: 5 + 5 = 10
+
+ collector.seal();
+
+ assertEquals(collector.getLengthOfShortestElement(), 1);
+ assertEquals(collector.getLengthOfLargestElement(), 5);
+ assertEquals(collector.getMaxRowLengthInBytes(), 10);
+ }
+
+ @Test
+ public void testBigDecimalLengthTracking() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.BIG_DECIMAL, true, false);
+
+ collector.collect(new BigDecimal("1.0"));
+ collector.collect(new BigDecimal("123456.789"));
+ collector.collect(new BigDecimal("99.99"));
+
+ collector.seal();
+
+ // BigDecimal tracks length
+ assertTrue(collector.getLengthOfShortestElement() > 0);
+ assertTrue(collector.getLengthOfLargestElement() > 0);
+ assertTrue(collector.getLengthOfLargestElement() >= collector.getLengthOfShortestElement());
+ }
+
+ @Test(expectedExceptions = UnsupportedOperationException.class)
+ public void testBigDecimalMultiValueNotSupported() {
+ AbstractColumnStatisticsCollector collector = createCollector(FieldSpec.DataType.BIG_DECIMAL, true, false);
+
+ // BigDecimal does not support multi-value
+ collector.collect((Object) new Object[]{new BigDecimal("1.0"), new BigDecimal("2.0")});
+ }
+
+ // Helper methods for test data generation
+
+ private Object[] getTestDataForType(FieldSpec.DataType dataType, boolean sorted) {
+ switch (dataType) {
+ case INT:
+ return sorted ? new Object[]{1, 5, 5, 10} : new Object[]{10, 1, 5, 5};
+ case LONG:
+ return sorted ? new Object[]{1L, 5L, 5L, 10L} : new Object[]{10L, 1L, 5L, 5L};
+ case FLOAT:
+ return sorted ? new Object[]{1.5f, 5.5f, 5.5f, 10.5f} : new Object[]{10.5f, 1.5f, 5.5f, 5.5f};
+ case DOUBLE:
+ return sorted ? new Object[]{1.5, 5.5, 5.5, 10.5} : new Object[]{10.5, 1.5, 5.5, 5.5};
+ case STRING:
+ return sorted ? new Object[]{"a", "b", "b", "c"} : new Object[]{"c", "a", "b", "b"};
+ case BYTES:
+ return sorted ? new Object[]{
+ new byte[]{1}, new byte[]{2}, new byte[]{2}, new byte[]{3}
+ } : new Object[]{
+ new byte[]{3}, new byte[]{1}, new byte[]{2}, new byte[]{2}
+ };
+ case BIG_DECIMAL:
+ return sorted ? new Object[]{
+ new BigDecimal("1.0"), new BigDecimal("5.0"), new BigDecimal("5.0"), new BigDecimal("10.0")
+ } : new Object[]{
+ new BigDecimal("10.0"), new BigDecimal("1.0"), new BigDecimal("5.0"), new BigDecimal("5.0")
+ };
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object[][] getMultiValueTestData(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return new Object[][]{
+ {1, 2},
+ {3, 4, 5}
+ };
+ case LONG:
+ return new Object[][]{
+ {1L, 2L},
+ {3L, 4L, 5L}
+ };
+ case FLOAT:
+ return new Object[][]{
+ {1.5f, 2.5f},
+ {3.5f, 4.5f, 5.5f}
+ };
+ case DOUBLE:
+ return new Object[][]{
+ {1.5, 2.5},
+ {3.5, 4.5, 5.5}
+ };
+ case STRING:
+ return new Object[][]{
+ {"a", "b"},
+ {"c", "d", "e"}
+ };
+ case BYTES:
+ return new Object[][]{
+ {new byte[]{1}, new byte[]{2}},
+ {new byte[]{3}, new byte[]{4}, new byte[]{5}}
+ };
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object getSingleValueForType(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return 42;
+ case LONG:
+ return 42L;
+ case FLOAT:
+ return 42.5f;
+ case DOUBLE:
+ return 42.5;
+ case STRING:
+ return "test";
+ case BYTES:
+ return new byte[]{1, 2, 3};
+ case BIG_DECIMAL:
+ return new BigDecimal("42.5");
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object getMinValue(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return 1;
+ case LONG:
+ return 1L;
+ case FLOAT:
+ return 1.5f;
+ case DOUBLE:
+ return 1.5;
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object getMaxValue(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return 10;
+ case LONG:
+ return 10L;
+ case FLOAT:
+ return 10.5f;
+ case DOUBLE:
+ return 10.5;
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object getExpectedMinForType(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return 1;
+ case LONG:
+ return 1L;
+ case FLOAT:
+ return 1.5f;
+ case DOUBLE:
+ return 1.5;
+ case STRING:
+ return "a";
+ case BYTES:
+ return new ByteArray(new byte[]{1});
+ case BIG_DECIMAL:
+ return new BigDecimal("1.0");
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private Object getExpectedMaxForType(FieldSpec.DataType dataType) {
+ switch (dataType) {
+ case INT:
+ return 10;
+ case LONG:
+ return 10L;
+ case FLOAT:
+ return 10.5f;
+ case DOUBLE:
+ return 10.5;
+ case STRING:
+ return "c";
+ case BYTES:
+ return new ByteArray(new byte[]{3});
+ case BIG_DECIMAL:
+ return new BigDecimal("10.0");
+ default:
+ throw new IllegalArgumentException("Unsupported data type: " + dataType);
+ }
+ }
+
+ private int getArrayLength(Object array) {
+ if (array instanceof int[]) {
+ return ((int[]) array).length;
+ } else if (array instanceof long[]) {
+ return ((long[]) array).length;
+ } else if (array instanceof float[]) {
+ return ((float[]) array).length;
+ } else if (array instanceof double[]) {
+ return ((double[]) array).length;
+ } else if (array instanceof Object[]) {
+ return ((Object[]) array).length;
+ }
+ throw new IllegalArgumentException("Unsupported array type: " + array.getClass());
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReaderTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReaderTest.java
new file mode 100644
index 0000000000..abe05a48af
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/readers/DefaultValueColumnReaderTest.java
@@ -0,0 +1,619 @@
+/**
+ * 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.segment.local.segment.readers;
+
+import java.io.IOException;
+import java.util.Arrays;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.MetricFieldSpec;
+import org.apache.pinot.spi.data.readers.MultiValueResult;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Comprehensive tests for DefaultValueColumnReader.
+ *
+ *
This test validates:
+ *
+ *
Single-value fields for all data types (INT, LONG, FLOAT, DOUBLE, STRING, BYTES)