You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2025/12/10 01:39:36 UTC
(pinot) branch master updated: Refactor segment creator and finalizer to make it reusable across different ways of indexing data (#17191)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new f4352e7bdfb Refactor segment creator and finalizer to make it reusable across different ways of indexing data (#17191)
f4352e7bdfb is described below
commit f4352e7bdfb842c7f9faf99b1b4fc7aa10872346
Author: Krishan Goyal <kr...@startree.ai>
AuthorDate: Wed Dec 10 07:09:32 2025 +0530
Refactor segment creator and finalizer to make it reusable across different ways of indexing data (#17191)
---
.../segment/processing/mapper/SegmentMapper.java | 10 +-
.../processing/partitioner/PartitionerFactory.java | 15 +
.../BaseSegmentProcessorFrameworkTest.java | 1252 ++++++++++++++++++++
.../framework/SegmentProcessorFrameworkTest.java | 711 ++---------
...arIndexCreator.java => BaseSegmentCreator.java} | 827 +++++++------
.../segment/creator/impl/ColumnIndexCreators.java | 120 ++
.../creator/impl/SegmentColumnarIndexCreator.java | 687 +----------
.../impl/SegmentIndexCreationDriverImpl.java | 225 +---
.../ColumnarSegmentPreIndexStatsContainer.java | 8 +-
.../stats/SegmentPreIndexStatsCollectorImpl.java | 16 +-
.../creator/impl/stats/StatsCollectorUtil.java | 14 +-
.../pinot/segment/spi/creator/SegmentCreator.java | 27 +-
12 files changed, 2020 insertions(+), 1892 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/mapper/SegmentMapper.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/mapper/SegmentMapper.java
index 49a60c52164..693df8615ec 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/mapper/SegmentMapper.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/mapper/SegmentMapper.java
@@ -35,7 +35,6 @@ import org.apache.pinot.core.segment.processing.genericrow.AdaptiveSizeBasedWrit
import org.apache.pinot.core.segment.processing.genericrow.FileWriter;
import org.apache.pinot.core.segment.processing.genericrow.GenericRowFileManager;
import org.apache.pinot.core.segment.processing.partitioner.Partitioner;
-import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
import org.apache.pinot.core.segment.processing.timehandler.TimeHandler;
import org.apache.pinot.core.segment.processing.timehandler.TimeHandlerFactory;
@@ -114,14 +113,9 @@ public class SegmentMapper {
schema.isEnableColumnBasedNullHandling() || tableConfig.getIndexingConfig().isNullHandlingEnabled();
_transformPipeline = transformPipeline;
_timeHandler = TimeHandlerFactory.getTimeHandler(processorConfig);
- List<PartitionerConfig> partitionerConfigs = processorConfig.getPartitionerConfigs();
- int numPartitioners = partitionerConfigs.size();
- _partitioners = new Partitioner[numPartitioners];
- for (int i = 0; i < numPartitioners; i++) {
- _partitioners[i] = PartitionerFactory.getPartitioner(partitionerConfigs.get(i));
- }
+ _partitioners = PartitionerFactory.getPartitioners(processorConfig.getPartitionerConfigs());
// Time partition + partition from partitioners
- _partitionsBuffer = new String[numPartitioners + 1];
+ _partitionsBuffer = new String[_partitioners.length + 1];
LOGGER.info("Initialized mapper with {} record readers, output dir: {}, timeHandler: {}, partitioners: {}",
_recordReaderFileConfigs.size(), _mapperOutputDir, _timeHandler.getClass(),
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/partitioner/PartitionerFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/partitioner/PartitionerFactory.java
index 40814e21a42..969aa4cd966 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/partitioner/PartitionerFactory.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/processing/partitioner/PartitionerFactory.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.segment.processing.partitioner;
import com.google.common.base.Preconditions;
+import java.util.List;
/**
@@ -85,4 +86,18 @@ public final class PartitionerFactory {
}
return partitioner;
}
+
+ /**
+ * Create partitioner array from configuration.
+ *
+ * @return Array of partitioners
+ */
+ public static Partitioner[] getPartitioners(List<PartitionerConfig> partitionerConfigs) {
+ int numPartitioners = partitionerConfigs.size();
+ Partitioner[] partitioners = new Partitioner[numPartitioners];
+ for (int i = 0; i < numPartitioners; i++) {
+ partitioners[i] = PartitionerFactory.getPartitioner(partitionerConfigs.get(i));
+ }
+ return partitioners;
+ }
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/BaseSegmentProcessorFrameworkTest.java b/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/BaseSegmentProcessorFrameworkTest.java
new file mode 100644
index 00000000000..a989c2acc1a
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/BaseSegmentProcessorFrameworkTest.java
@@ -0,0 +1,1252 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.segment.processing.framework;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.IntStream;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.timehandler.TimeHandler;
+import org.apache.pinot.core.segment.processing.timehandler.TimeHandlerConfig;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.ComplexTypeConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+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.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/**
+ * Abstract base class for SegmentProcessorFramework tests.
+ * Contains all test logic and setup, with abstract methods for processing that
+ * concrete implementations override to use either SPF or CSPF.
+ */
+public abstract class BaseSegmentProcessorFrameworkTest {
+ protected static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "SegmentProcessorFrameworkTest");
+
+ protected List<File> _singleSegment;
+ protected List<File> _multipleSegments;
+ protected List<File> _multiValueSegments;
+
+ protected TableConfig _tableConfig;
+ protected TableConfig _tableConfigWithoutTimeColumn;
+ protected TableConfig _tableConfigNullValueEnabled;
+ protected TableConfig _tableConfigSegmentNameGeneratorEnabled;
+ protected TableConfig _tableConfigWithFixedSegmentName;
+
+ protected Schema _schema;
+ protected Schema _schemaMV;
+ protected Schema _schemaWithComplexType;
+
+ private final List<Object[]> _rawData =
+ Arrays.asList(new Object[]{"abc", 1000, 1597719600000L}, new Object[]{null, 2000, 1597773600000L},
+ new Object[]{"abc", null, 1597777200000L}, new Object[]{"abc", 4000, 1597795200000L},
+ new Object[]{"abc", 3000, 1597802400000L}, new Object[]{null, null, 1597838400000L},
+ new Object[]{"xyz", 4000, 1597856400000L}, new Object[]{null, 1000, 1597878000000L},
+ new Object[]{"abc", 7000, 1597881600000L}, new Object[]{"xyz", 6000, 1597892400000L});
+
+ private final List<Object[]> _rawDataMultiValue =
+ Arrays.asList(new Object[]{new String[]{"a", "b"}, 1000, 1597795200000L},
+ new Object[]{null, null, 1597795200000L}, new Object[]{null, 1000, 1597795200000L},
+ new Object[]{new String[]{"a", "b"}, null, 1597795200000L});
+
+ /**
+ * Abstract method for processing segments.
+ * Concrete implementations create appropriate readers (RecordReader for SPF, ColumnarDataSource for CSPF).
+ */
+ protected abstract List<File> processSegments(List<File> segmentDirs, SegmentProcessorConfig config, File workingDir)
+ throws Exception;
+
+ @BeforeClass
+ public void setup()
+ throws Exception {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ FileUtils.forceMkdir(TEMP_DIR);
+
+ _tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
+ _tableConfigNullValueEnabled =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time")
+ .setNullHandlingEnabled(true).build();
+ _tableConfigWithoutTimeColumn =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setNullHandlingEnabled(true).build();
+ _tableConfigSegmentNameGeneratorEnabled =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
+ _tableConfigSegmentNameGeneratorEnabled.getIndexingConfig().setSegmentNameGeneratorType("normalizedDate");
+ _tableConfigWithFixedSegmentName =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
+ _tableConfigWithFixedSegmentName.getIndexingConfig().setSegmentNameGeneratorType("fixed");
+
+ _schema =
+ new Schema.SchemaBuilder().setSchemaName("mySchema")
+ .addSingleValueDimension("campaign", DataType.STRING, "")
+ .addSingleValueDimension("campaign.inner1", DataType.STRING)
+ .addSingleValueDimension("campaign.inner1.inner2", DataType.STRING)
+ // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
+ .addMetric("clicks", DataType.INT, 1000)
+ .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ _schemaMV =
+ new Schema.SchemaBuilder().setSchemaName("mySchema")
+ .addMultiValueDimension("campaign", DataType.STRING, "")
+ // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
+ .addMetric("clicks", DataType.INT, 1000)
+ .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ _schemaWithComplexType =
+ new Schema.SchemaBuilder().setSchemaName("mySchema")
+ .addSingleValueDimension("campaign", DataType.JSON)
+ .addSingleValueDimension("campaign.inner1", DataType.STRING)
+ .addSingleValueDimension("campaign.inner1.inner2", DataType.STRING)
+ .addSingleValueDimension("targetusers.user", DataType.STRING)
+ // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
+ .addMetric("clicks", DataType.INT, 1000)
+ .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ // create segments in many folders
+ _singleSegment = createInputSegments(new File(TEMP_DIR, "single_segment"), _rawData, 1, _schema);
+ _multipleSegments = createInputSegments(new File(TEMP_DIR, "multiple_segments"), _rawData, 3, _schema);
+ _multiValueSegments =
+ createInputSegments(new File(TEMP_DIR, "multi_value_segment"), _rawDataMultiValue, 1, _schemaMV);
+ }
+
+ /**
+ * Create input segments and return the list of segment directories.
+ * This is changed from the original to return List<File> instead of List<RecordReader>.
+ */
+ private List<File> createInputSegments(File inputDir, List<Object[]> rawData, int numSegments, Schema schema)
+ throws Exception {
+ assertTrue(inputDir.mkdirs());
+
+ List<List<GenericRow>> dataLists = new ArrayList<>();
+ if (numSegments > 1) {
+ List<GenericRow> dataList1 = new ArrayList<>(4);
+ IntStream.range(0, 4).forEach(i -> dataList1.add(getGenericRow(rawData.get(i))));
+ dataLists.add(dataList1);
+ List<GenericRow> dataList2 = new ArrayList<>(4);
+ IntStream.range(4, 7).forEach(i -> dataList2.add(getGenericRow(rawData.get(i))));
+ dataLists.add(dataList2);
+ List<GenericRow> dataList3 = new ArrayList<>(4);
+ IntStream.range(7, 10).forEach(i -> dataList3.add(getGenericRow(rawData.get(i))));
+ dataLists.add(dataList3);
+ } else {
+ List<GenericRow> dataList = new ArrayList<>();
+ rawData.forEach(r -> dataList.add(getGenericRow(r)));
+ dataLists.add(dataList);
+ }
+
+ List<File> segmentDirs = new ArrayList<>(dataLists.size());
+ int idx = 0;
+ for (List<GenericRow> inputRows : dataLists) {
+ RecordReader recordReader = new GenericRowRecordReader(inputRows);
+ // NOTE: Generate segments with null value enabled
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(_tableConfigNullValueEnabled, schema);
+ segmentGeneratorConfig.setOutDir(inputDir.getPath());
+ segmentGeneratorConfig.setSequenceId(idx++);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, recordReader);
+ driver.build();
+ segmentDirs.add(driver.getOutputDirectory());
+ }
+ return segmentDirs;
+ }
+
+ private GenericRow getGenericRow(Object[] rawRow) {
+ GenericRow row = new GenericRow();
+ row.putValue("campaign", rawRow[0]);
+ row.putValue("clicks", rawRow[1]);
+ row.putValue("time", rawRow[2]);
+ return row;
+ }
+
+ @Test
+ public void testSegmentGenerationWithComplexType() throws Exception {
+ File workingDir = new File(TEMP_DIR, "single_segment_complex_type_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create segments for complex type test
+ GenericRow genericRow = new GenericRow();
+ genericRow.putValue("a", 1L);
+ Map<String, Object> map1 = new HashMap<>();
+ genericRow.putValue("campaign", map1);
+ map1.put("inner", "innerv");
+ Map<String, Object> innerMap1 = new HashMap<>();
+ innerMap1.put("inner2", "inner2v");
+
+ map1.put("inner1", innerMap1);
+ Map<String, Object> map2 = new HashMap<>();
+ map2.put("c", 3);
+ genericRow.putValue("map2", map2);
+
+ //list with two map entries inside
+ List<Map<String, Object>> list = new ArrayList<>();
+ Map<String, Object> map3 = new HashMap<>();
+ map3.put("user", "foobar");
+ list.add(map3);
+ Map<String, Object> map4 = new HashMap<>();
+ map4.put("user", "barfoo");
+ list.add(map4);
+ genericRow.putValue("targetusers", list);
+
+ List<GenericRow> inputRows = List.of(genericRow);
+
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ List<String> fieldsToUnnest = new ArrayList<>();
+ fieldsToUnnest.add("targetusers");
+
+ ingestionConfig.setComplexTypeConfig(
+ new ComplexTypeConfig(fieldsToUnnest, ".", null, null));
+ _tableConfig.setIngestionConfig(ingestionConfig);
+
+ // Create a segment with this data
+ File complexTypeInputDir = new File(TEMP_DIR, "complex_type_input");
+ FileUtils.forceMkdir(complexTypeInputDir);
+ RecordReader recordReader = new GenericRowRecordReader(inputRows);
+ SegmentGeneratorConfig segmentGeneratorConfig =
+ new SegmentGeneratorConfig(_tableConfig, _schemaWithComplexType);
+ segmentGeneratorConfig.setOutDir(complexTypeInputDir.getPath());
+ segmentGeneratorConfig.setSequenceId(0);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, recordReader);
+ driver.build();
+ List<File> segmentDirs = List.of(driver.getOutputDirectory());
+
+ // Default configs
+ SegmentProcessorConfig config =
+ new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schemaWithComplexType).build();
+
+ List<File> outputSegments = processSegments(segmentDirs, config, workingDir);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+ // Pick the column created from complex type
+ ColumnMetadata campaignInner2Metadata = segmentMetadata.getColumnMetadataFor("campaign.inner1.inner2");
+ // Verify we see a specific value parsed from the complexType
+ Assert.assertEquals(campaignInner2Metadata.getMinValue().compareTo("inner2v"), 0);
+ ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
+ Assert.assertEquals(
+ campaignMetadata.getMinValue().compareTo("{\"inner1\":{\"inner2\":\"inner2v\"},\"inner\":\"innerv\"}"), 0);
+
+ ColumnMetadata listMetadata = segmentMetadata.getColumnMetadataFor("targetusers.user");
+ Assert.assertEquals(listMetadata.getMinValue().compareTo("barfoo"), 0);
+ }
+
+ @Test
+ public void testSingleSegment()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "single_segment_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Default configs
+ SegmentProcessorConfig config =
+ new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
+ List<File> outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+ assertEquals(segmentMetadata.getTotalDocs(), 10);
+ ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
+ assertEquals(campaignMetadata.getCardinality(), 3);
+ assertEquals(campaignMetadata.getMinValue(), "");
+ assertEquals(campaignMetadata.getMaxValue(), "xyz");
+ ColumnMetadata clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
+ assertEquals(clicksMetadata.getCardinality(), 6);
+ assertEquals(clicksMetadata.getMinValue(), 1000);
+ assertEquals(clicksMetadata.getMaxValue(), 7000);
+ ColumnMetadata timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 10);
+ assertEquals(timeMetadata.getMinValue(), 1597719600000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
+ DataSource campaignDataSource = segment.getDataSource("campaign");
+ assertNull(campaignDataSource.getNullValueVector());
+ DataSource clicksDataSource = segment.getDataSource("clicks");
+ assertNull(clicksDataSource.getNullValueVector());
+ DataSource timeDataSource = segment.getDataSource("time");
+ assertNull(timeDataSource.getNullValueVector());
+ assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
+ segment.destroy();
+ FileUtils.cleanDirectory(workingDir);
+
+ // Default configs - null value enabled
+ config =
+ new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigNullValueEnabled).setSchema(_schema).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ String[] outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ segmentMetadata = segment.getSegmentMetadata();
+ assertEquals(segmentMetadata.getTotalDocs(), 10);
+ campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
+ assertEquals(campaignMetadata.getCardinality(), 3);
+ assertEquals(campaignMetadata.getMinValue(), "");
+ assertEquals(campaignMetadata.getMaxValue(), "xyz");
+ clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
+ assertEquals(clicksMetadata.getCardinality(), 6);
+ assertEquals(clicksMetadata.getMinValue(), 1000);
+ assertEquals(clicksMetadata.getMaxValue(), 7000);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 10);
+ assertEquals(timeMetadata.getMinValue(), 1597719600000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
+ campaignDataSource = segment.getDataSource("campaign");
+ NullValueVectorReader campaignNullValueVector = campaignDataSource.getNullValueVector();
+ assertNotNull(campaignNullValueVector);
+ assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{1, 5, 7});
+ clicksDataSource = segment.getDataSource("clicks");
+ NullValueVectorReader clicksNullValueVector = clicksDataSource.getNullValueVector();
+ assertNotNull(clicksNullValueVector);
+ assertEquals(clicksNullValueVector.getNullBitmap().toArray(), new int[]{2, 5});
+ timeDataSource = segment.getDataSource("time");
+ assertNull(timeDataSource.getNullValueVector());
+ assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
+ segment.destroy();
+ FileUtils.cleanDirectory(workingDir);
+
+ // Fixed segment name
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigWithFixedSegmentName).setSchema(_schema)
+ .setSegmentConfig(new SegmentConfig.Builder().setFixedSegmentName("myTable_segment_0001").build()).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ segmentMetadata = segment.getSegmentMetadata();
+ assertEquals(segmentMetadata.getName(), "myTable_segment_0001");
+ segment.destroy();
+ FileUtils.cleanDirectory(workingDir);
+
+ // Segment config
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setSegmentConfig(
+ new SegmentConfig.Builder().setMaxNumRecordsPerSegment(4).setSegmentNamePrefix("myPrefix")
+ .setSegmentNamePostfix("myPostfix").build()).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 3);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ outputSegments.sort(null);
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 4);
+ assertEquals(segmentMetadata.getName(), "myPrefix_1597719600000_1597795200000_myPostfix_0");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
+ assertEquals(segmentMetadata.getTotalDocs(), 4);
+ assertEquals(segmentMetadata.getName(), "myPrefix_1597802400000_1597878000000_myPostfix_1");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ assertEquals(segmentMetadata.getName(), "myPrefix_1597881600000_1597892400000_myPostfix_2");
+ FileUtils.cleanDirectory(workingDir);
+
+ config =
+ new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigSegmentNameGeneratorEnabled).setSchema(_schema)
+ .setSegmentConfig(new SegmentConfig.Builder().setMaxNumRecordsPerSegment(4).setSegmentNamePrefix("myPrefix")
+ .setSegmentNamePostfix("myPostfix").build()).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 3);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ outputSegments.sort(null);
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 4);
+ assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-18_2020-08-19_myPostfix_0");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
+ assertEquals(segmentMetadata.getTotalDocs(), 4);
+ assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-19_2020-08-19_myPostfix_1");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-20_2020-08-20_myPostfix_2");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testSingleSegmentWithTimeFiltersAndPartition()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "single_segment_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Time filter
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597795200000L, 1597881600000L).build())
+ .build();
+ List<File> outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ String[] outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ ColumnMetadata timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 5);
+ assertEquals(timeMetadata.getMinValue(), 1597795200000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597878000000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597878000000_0");
+ FileUtils.cleanDirectory(workingDir);
+
+ // Negate time filter
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597795200000L, 1597881600000L)
+ .setNegateWindowFilter(true).build()).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 5);
+ assertEquals(timeMetadata.getMinValue(), 1597719600000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
+ FileUtils.cleanDirectory(workingDir);
+
+ // Time filter - filtered everything
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597968000000L, 1598054400000L).build())
+ .build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertTrue(outputSegments.isEmpty());
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ FileUtils.cleanDirectory(workingDir);
+
+ // Time round
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema)
+ .setTimeHandlerConfig(new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).build())
+ .build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 10);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 3);
+ assertEquals(timeMetadata.getMinValue(), 1597708800000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597881600000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597881600000_0");
+ FileUtils.cleanDirectory(workingDir);
+
+ // Time partition
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setPartitionBucketMs(86400000).build()).build();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 3);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ outputSegments.sort(null);
+ // segment 0
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 3);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 3);
+ assertEquals(timeMetadata.getMinValue(), 1597719600000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597777200000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597777200000_0");
+ // segment 1
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 5);
+ assertEquals(timeMetadata.getMinValue(), 1597795200000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597878000000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597878000000_1");
+ // segment 2
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 2);
+ assertEquals(timeMetadata.getMinValue(), 1597881600000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
+ assertEquals(segmentMetadata.getName(), "myTable_1597881600000_1597892400000_2");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testSingleSegmentWithRollup()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "single_segment_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Time filter, round, partition, rollup - null value enabled
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfigNullValueEnabled).setSchema(_schema)
+ .setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597708800000L, 1597881600000L)
+ .setRoundBucketMs(86400000).setPartitionBucketMs(86400000).build()).setMergeType(MergeType.ROLLUP)
+ .build();
+ List<File> outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 2);
+ String[] outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ outputSegments.sort(null);
+ // segment 0
+ ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
+ assertEquals(campaignMetadata.getCardinality(), 2);
+ assertEquals(campaignMetadata.getMinValue(), "");
+ assertEquals(campaignMetadata.getMaxValue(), "abc");
+ ColumnMetadata clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
+ assertEquals(clicksMetadata.getCardinality(), 2);
+ assertEquals(clicksMetadata.getMinValue(), 1000);
+ assertEquals(clicksMetadata.getMaxValue(), 2000);
+ ColumnMetadata timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 1);
+ assertEquals(timeMetadata.getMinValue(), 1597708800000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597708800000L);
+ DataSource campaignDataSource = segment.getDataSource("campaign");
+ NullValueVectorReader campaignNullValueVector = campaignDataSource.getNullValueVector();
+ assertNotNull(campaignNullValueVector);
+ assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{0});
+ DataSource clicksDataSource = segment.getDataSource("clicks");
+ assertNull(clicksDataSource.getNullValueVector());
+ DataSource timeDataSource = segment.getDataSource("time");
+ assertNull(timeDataSource.getNullValueVector());
+ assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597708800000_0");
+ segment.destroy();
+ // segment 1
+ segment = ImmutableSegmentLoader.load(outputSegments.get(1), ReadMode.mmap);
+ segmentMetadata = segment.getSegmentMetadata();
+ assertEquals(segmentMetadata.getTotalDocs(), 3);
+ campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
+ assertEquals(campaignMetadata.getCardinality(), 3);
+ assertEquals(campaignMetadata.getMinValue(), "");
+ assertEquals(campaignMetadata.getMaxValue(), "xyz");
+ clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
+ assertEquals(clicksMetadata.getCardinality(), 3);
+ assertEquals(clicksMetadata.getMinValue(), 1000);
+ assertEquals(clicksMetadata.getMaxValue(), 7000);
+ timeMetadata = segmentMetadata.getColumnMetadataFor("time");
+ assertEquals(timeMetadata.getCardinality(), 1);
+ assertEquals(timeMetadata.getMinValue(), 1597795200000L);
+ assertEquals(timeMetadata.getMaxValue(), 1597795200000L);
+ timeDataSource = segment.getDataSource("campaign");
+ campaignNullValueVector = timeDataSource.getNullValueVector();
+ assertNotNull(campaignNullValueVector);
+ assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{0});
+ clicksDataSource = segment.getDataSource("clicks");
+ assertNull(clicksDataSource.getNullValueVector());
+ timeDataSource = segment.getDataSource("time");
+ assertNull(timeDataSource.getNullValueVector());
+ assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_1");
+ segment.destroy();
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testSingleSegmentWithDedup()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "single_segment_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Time round, dedup
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema)
+ .setTimeHandlerConfig(new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).build())
+ .setMergeType(MergeType.DEDUP).build();
+ List<File> outputSegments = processSegments(_singleSegment, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ String[] outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 8);
+ assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597881600000_0");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ /**
+ * Helper method to create multiple input segments with controlled data distribution.
+ */
+ protected List<File> createMultipleInputSegments(File inputDir, List<List<Object[]>> segmentDataLists,
+ Schema schema)
+ throws Exception {
+ assertTrue(inputDir.mkdirs());
+
+ List<File> segmentDirs = new ArrayList<>(segmentDataLists.size());
+ int idx = 0;
+ for (List<Object[]> segmentData : segmentDataLists) {
+ List<GenericRow> inputRows = new ArrayList<>();
+ for (Object[] rawRow : segmentData) {
+ inputRows.add(getGenericRow(rawRow));
+ }
+ RecordReader recordReader = new GenericRowRecordReader(inputRows);
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(_tableConfigNullValueEnabled, schema);
+ segmentGeneratorConfig.setOutDir(inputDir.getPath());
+ segmentGeneratorConfig.setSequenceId(idx++);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, recordReader);
+ driver.build();
+ segmentDirs.add(driver.getOutputDirectory());
+ }
+ return segmentDirs;
+ }
+
+ /**
+ * Helper method to create segment with invalid data for specific column/rows.
+ */
+ protected List<File> createSegmentWithInvalidData(File inputDir, List<Object[]> data, Schema schema,
+ String invalidColumn, Set<Integer> invalidRowIndices)
+ throws Exception {
+ assertTrue(inputDir.mkdirs());
+
+ List<GenericRow> inputRows = new ArrayList<>();
+ for (int i = 0; i < data.size(); i++) {
+ GenericRow row = new GenericRow();
+ Object[] rawRow = data.get(i);
+ row.putValue("campaign", rawRow[0]);
+
+ // Set invalid value for specified rows and column
+ if (invalidRowIndices.contains(i) && invalidColumn.equals("clicks")) {
+ row.putValue("clicks", "INVALID");
+ } else if (invalidRowIndices.contains(i) && invalidColumn.equals("time")) {
+ row.putValue("time", "INVALID_TIME");
+ } else {
+ row.putValue("clicks", rawRow[1]);
+ row.putValue("time", rawRow[2]);
+ }
+
+ inputRows.add(row);
+ }
+
+ // Add the invalid column as STRING type to accommodate invalid data in the segment
+ Schema schemaWithInvalidColumn = new Schema.SchemaBuilder().build();
+ for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
+ schemaWithInvalidColumn.addField(fieldSpec);
+ }
+ schemaWithInvalidColumn.removeField(invalidColumn);
+ schemaWithInvalidColumn.addField(new DimensionFieldSpec(invalidColumn, DataType.STRING, true));
+
+ RecordReader recordReader = new GenericRowRecordReader(inputRows);
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(_tableConfigWithoutTimeColumn,
+ schemaWithInvalidColumn);
+ segmentGeneratorConfig.setOutDir(inputDir.getPath());
+ segmentGeneratorConfig.setSequenceId(0);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, recordReader);
+ driver.build();
+ return List.of(driver.getOutputDirectory());
+ }
+
+ /**
+ * Helper method to generate test data with specific partition values and times.
+ */
+ protected List<Object[]> generatePartitionedData(int numDocs, String campaignValue, long baseTime) {
+ List<Object[]> data = new ArrayList<>();
+ for (int i = 0; i < numDocs; i++) {
+ data.add(new Object[]{campaignValue, 1000 + i, baseTime + i * 1000L});
+ }
+ return data;
+ }
+
+ @Test
+ public void testMultipleDataSourcesWithPartitioningAndNulls()
+ throws Exception {
+ Schema schemaWithDefault = new Schema.SchemaBuilder().build();
+ for (FieldSpec fieldSpec : _schema.getAllFieldSpecs()) {
+ schemaWithDefault.addField(fieldSpec);
+ }
+ schemaWithDefault.removeField("campaign");
+ schemaWithDefault.addField(new DimensionFieldSpec("campaign", DataType.STRING, true));
+ testMultipleDataSourcesWithPartitioningAndNulls(_tableConfigNullValueEnabled, schemaWithDefault, 12);
+ }
+
+ @Test
+ public void testMultipleDataSourcesWithPartitioningAndDefaults()
+ throws Exception {
+ Schema schemaWithDefault = new Schema.SchemaBuilder().build();
+ for (FieldSpec fieldSpec : _schema.getAllFieldSpecs()) {
+ schemaWithDefault.addField(fieldSpec);
+ }
+ schemaWithDefault.removeField("campaign");
+ schemaWithDefault.addField(new DimensionFieldSpec("campaign", DataType.STRING, true, "abc"));
+ testMultipleDataSourcesWithPartitioningAndNulls(_tableConfigNullValueEnabled, schemaWithDefault, 9);
+ }
+
+ private void testMultipleDataSourcesWithPartitioningAndNulls(TableConfig tableConf, Schema schema, int numOfSegments)
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "multiple_datasources_partitioning_output" + UUID.randomUUID());
+ FileUtils.forceMkdir(workingDir);
+
+ // Create 3 input segments with different sizes: 100, 50, 200 docs
+ // Data spans 3 days and has multiple campaign values (3 distinct + nulls)
+ long day1 = 1597708800000L; // 2020-08-18
+ long day2 = 1597795200000L; // 2020-08-19
+ long day3 = 1597881600000L; // 2020-08-20
+
+ List<List<Object[]>> segmentDataLists = new ArrayList<>();
+
+ // Segment 1: 100 docs
+ List<Object[]> segment1Data = new ArrayList<>();
+ for (int i = 0; i < 30; i++) {
+ segment1Data.add(new Object[]{"abc", 1000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 25; i++) {
+ segment1Data.add(new Object[]{"xyz", 2000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 20; i++) {
+ segment1Data.add(new Object[]{"abc", 3000 + i, day2 + i * 1000});
+ }
+ for (int i = 0; i < 15; i++) {
+ segment1Data.add(new Object[]{null, 4000 + i, day2 + i * 1000});
+ }
+ for (int i = 0; i < 10; i++) {
+ segment1Data.add(new Object[]{"pqr", null, day3 + i * 1000});
+ }
+ // Shuffle to mix data
+ Collections.shuffle(segment1Data, new Random());
+
+ // Segment 2: 50 docs
+ List<Object[]> segment2Data = new ArrayList<>();
+ for (int i = 0; i < 15; i++) {
+ segment2Data.add(new Object[]{"abc", 5000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 10; i++) {
+ segment2Data.add(new Object[]{"xyz", 6000 + i, day2 + i * 1000});
+ }
+ for (int i = 0; i < 12; i++) {
+ segment2Data.add(new Object[]{"pqr", 7000 + i, day2 + i * 1000});
+ }
+ for (int i = 0; i < 8; i++) {
+ segment2Data.add(new Object[]{null, null, day1 + i * 1000});
+ }
+ for (int i = 0; i < 5; i++) {
+ segment2Data.add(new Object[]{"abc", 8000 + i, day3 + i * 1000});
+ }
+ Collections.shuffle(segment2Data, new Random());
+
+ // Segment 3: 200 docs
+ List<Object[]> segment3Data = new ArrayList<>();
+ for (int i = 0; i < 50; i++) {
+ segment3Data.add(new Object[]{"abc", 9000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 40; i++) {
+ segment3Data.add(new Object[]{"xyz", 10000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 35; i++) {
+ segment3Data.add(new Object[]{"pqr", 11000 + i, day1 + i * 1000});
+ }
+ for (int i = 0; i < 30; i++) {
+ segment3Data.add(new Object[]{"abc", null, day2 + i * 1000});
+ }
+ for (int i = 0; i < 25; i++) {
+ segment3Data.add(new Object[]{null, 12000 + i, day3 + i * 1000});
+ }
+ for (int i = 0; i < 20; i++) {
+ segment3Data.add(new Object[]{"xyz", 13000 + i, day3 + i * 1000});
+ }
+ Collections.shuffle(segment3Data, new Random());
+
+ segmentDataLists.add(segment1Data);
+ segmentDataLists.add(segment2Data);
+ segmentDataLists.add(segment3Data);
+
+ File inputDir = new File(TEMP_DIR, "multiple_datasources_partitioning_input" + UUID.randomUUID());
+ List<File> inputSegments = createMultipleInputSegments(inputDir, segmentDataLists, schema);
+
+ // Configure time + column partitioning
+ PartitionerConfig columnPartitioner = new PartitionerConfig.Builder()
+ .setPartitionerType(PartitionerFactory.PartitionerType.COLUMN_VALUE)
+ .setColumnName("campaign")
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(tableConf)
+ .setSchema(schema)
+ .setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH)
+ .setPartitionBucketMs(86400000) // 1 day
+ .build())
+ .setPartitionerConfigs(List.of(columnPartitioner))
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify output
+ assertEquals(outputSegments.size(), numOfSegments, "Should create output segments");
+
+ // Total docs should be 350 (100 + 50 + 200)
+ int totalDocs = 0;
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ totalDocs += metadata.getTotalDocs();
+ }
+ assertEquals(totalDocs, 350, "Total document count should match sum of inputs");
+
+ // Verify each segment has data for single day + single campaign
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ ColumnMetadata timeMetadata = metadata.getColumnMetadataFor("time");
+ assertNotNull(timeMetadata, "Time column metadata should not be null");
+ long startTime = (long) timeMetadata.getMinValue();
+ long endTime = (long) timeMetadata.getMaxValue();
+ // Verify time range is exactly 1 day
+ assertTrue(endTime - startTime <= 86400000, "Segment time range should be 1 day");
+
+ ColumnMetadata campaignMetadata = metadata.getColumnMetadataFor("campaign");
+ assertNotNull(campaignMetadata, "Campaign column metadata should not be null");
+ assertTrue(campaignMetadata.getCardinality() <= 1, "Campaign cardinality should be at most 1");
+ }
+
+ FileUtils.cleanDirectory(workingDir);
+ FileUtils.cleanDirectory(inputDir);
+ }
+
+ @Test
+ public void testSegmentSplittingWithinPartition()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "segment_splitting_within_partition_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create input segment with 100 docs, all same partition value
+ List<Object[]> testData = generatePartitionedData(100, "abc", 1597719600000L);
+ File inputDir = new File(TEMP_DIR, "segment_splitting_within_partition_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure max 30 records per segment
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setSegmentConfig(new SegmentConfig.Builder().setMaxNumRecordsPerSegment(30).build())
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify exactly 4 segments created (30+30+30+10)
+ assertEquals(outputSegments.size(), 4, "Should create exactly 4 segments");
+
+ outputSegments.sort(null);
+ SegmentMetadata seg0 = new SegmentMetadataImpl(outputSegments.get(0));
+ SegmentMetadata seg1 = new SegmentMetadataImpl(outputSegments.get(1));
+ SegmentMetadata seg2 = new SegmentMetadataImpl(outputSegments.get(2));
+ SegmentMetadata seg3 = new SegmentMetadataImpl(outputSegments.get(3));
+
+ assertEquals(seg0.getTotalDocs(), 30, "Segment 0 should have 30 docs");
+ assertEquals(seg1.getTotalDocs(), 30, "Segment 1 should have 30 docs");
+ assertEquals(seg2.getTotalDocs(), 30, "Segment 2 should have 30 docs");
+ assertEquals(seg3.getTotalDocs(), 10, "Segment 3 should have 10 docs");
+
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testSegmentSplittingAcrossMultiplePartitions()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "segment_splitting_across_partitions_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create input with 3 partitions: abc(80), xyz(70), pqr(45) = 195 docs total
+ List<Object[]> testData = new ArrayList<>();
+ testData.addAll(generatePartitionedData(80, "abc", 1597719600000L));
+ testData.addAll(generatePartitionedData(70, "xyz", 1597719600000L));
+ testData.addAll(generatePartitionedData(45, "pqr", 1597719600000L));
+
+ File inputDir = new File(TEMP_DIR, "segment_splitting_across_partitions_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure column partitioner + max 30 records per segment
+ PartitionerConfig columnPartitioner = new PartitionerConfig.Builder()
+ .setPartitionerType(PartitionerFactory.PartitionerType.COLUMN_VALUE)
+ .setColumnName("campaign")
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setSegmentConfig(new SegmentConfig.Builder().setMaxNumRecordsPerSegment(30).build())
+ .setPartitionerConfigs(List.of(columnPartitioner))
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify total 8 segments: abc(3), xyz(3), pqr(2)
+ assertEquals(outputSegments.size(), 8, "Should create exactly 8 segments");
+
+ // Verify total doc count
+ int totalDocs = 0;
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ totalDocs += metadata.getTotalDocs();
+ }
+ assertEquals(totalDocs, 195, "Total docs should be 195");
+
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testExactBoundaryCondition()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "exact_boundary_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create input with exactly 90 docs, all same partition
+ List<Object[]> testData = generatePartitionedData(90, "abc", 1597719600000L);
+ File inputDir = new File(TEMP_DIR, "exact_boundary_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure max 30 records per segment
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setSegmentConfig(new SegmentConfig.Builder().setMaxNumRecordsPerSegment(30).build())
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify exactly 3 segments (30+30+30), no empty 4th segment
+ assertEquals(outputSegments.size(), 3, "Should create exactly 3 segments");
+
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ assertEquals(metadata.getTotalDocs(), 30, "Each segment should have exactly 30 docs");
+ }
+
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testContinueOnErrorDuringPartitionMapping()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "continue_on_error_partition_mapping_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create data with 10 rows, rows 2, 5, 7 have invalid time
+ List<Object[]> testData = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ testData.add(new Object[]{"abc", 1000 + i, 1597719600000L + i * 1000});
+ }
+
+ File inputDir = new File(TEMP_DIR, "continue_on_error_partition_mapping_input");
+ Set<Integer> invalidRowIndices = new HashSet<>(Arrays.asList(2, 5, 7));
+ List<File> inputSegments = createSegmentWithInvalidData(inputDir, testData, _schema, "time", invalidRowIndices);
+
+ // Configure continueOnError = true with time partitioning
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setContinueOnError(true);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName("myTable")
+ .setTimeColumnName("time")
+ .setIngestionConfig(ingestionConfig)
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(tableConfig)
+ .setSchema(_schema)
+ .setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH)
+ .setPartitionBucketMs(86400000)
+ .build())
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ int totalDocs = 0;
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ totalDocs += metadata.getTotalDocs();
+ }
+ assertEquals(totalDocs, 10, "Should have 10 valid docs due to continue on error");
+
+ FileUtils.cleanDirectory(workingDir);
+
+ // Now create config with continueOnError = false and verify exception
+ tableConfig.setIngestionConfig(new IngestionConfig());
+ SegmentProcessorConfig configFailFast = new SegmentProcessorConfig.Builder()
+ .setTableConfig(tableConfig)
+ .setSchema(_schema)
+ .setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH)
+ .setPartitionBucketMs(86400000)
+ .build())
+ .build();
+ boolean exceptionThrown = false;
+ try {
+ processSegments(inputSegments, configFailFast, workingDir);
+ } catch (Exception e) {
+ exceptionThrown = true;
+ }
+ assertTrue(exceptionThrown, "Processing should throw exception with continueOnError=false");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testContinueOnErrorDuringStatsAndIndexing()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "continue_on_error_stats_indexing_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create data with 10 rows, rows 3, 5, 8 have invalid clicks value
+ List<Object[]> testData = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ testData.add(new Object[]{"abc", 1000 + i, 1597719600000L + i * 1000});
+ }
+
+ File inputDir = new File(TEMP_DIR, "continue_on_error_stats_indexing_input");
+ Set<Integer> invalidRowIndices = new HashSet<>(Arrays.asList(3, 5, 8));
+ List<File> inputSegments = createSegmentWithInvalidData(inputDir, testData, _schema, "clicks", invalidRowIndices);
+
+ // Configure continueOnError = true
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setContinueOnError(true);
+ TableConfig tableConfigWithContinueOnError = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName("myTable")
+ .setTimeColumnName("time")
+ .setIngestionConfig(ingestionConfig)
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(tableConfigWithContinueOnError)
+ .setSchema(_schema)
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify processing completes with all 10 docs
+ assertEquals(outputSegments.size(), 1, "Should create 1 output segment");
+
+ ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
+ SegmentMetadata metadata = segment.getSegmentMetadata();
+ assertEquals(metadata.getTotalDocs(), 10, "Should have all 10 docs");
+
+ // Verify rows with invalid values use default (1000)
+ try (PinotSegmentColumnReader clicksReader = new PinotSegmentColumnReader(segment, "clicks")) {
+ for (int i = 0; i < 10; i++) {
+ int clicksValue = (int) clicksReader.getValue(i);
+ if (invalidRowIndices.contains(i)) {
+ assertEquals(clicksValue, 1000, "Invalid rows should use default clicks value");
+ } else {
+ assertEquals(clicksValue, 1000 + i, "Valid rows should retain original clicks value");
+ }
+ }
+ }
+
+ segment.destroy();
+ FileUtils.cleanDirectory(workingDir);
+
+ // Now create config with continueOnError = false and verify exception
+ tableConfigWithContinueOnError.setIngestionConfig(new IngestionConfig());
+ SegmentProcessorConfig configFailFast = new SegmentProcessorConfig.Builder()
+ .setTableConfig(tableConfigWithContinueOnError)
+ .setSchema(_schema)
+ .build();
+ boolean exceptionThrown = false;
+ try {
+ processSegments(inputSegments, configFailFast, workingDir);
+ } catch (Exception e) {
+ exceptionThrown = true;
+ }
+ assertTrue(exceptionThrown, "Processing should throw exception with continueOnError=false");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testAllRecordsFilteredByTime()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "all_records_filtered_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create segment with times in 2020
+ List<Object[]> testData = generatePartitionedData(10, "abc", 1597719600000L);
+ File inputDir = new File(TEMP_DIR, "all_records_filtered_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure time filter for far future (2030-2035)
+ long futureStart = 1893456000000L; // Jan 1, 2030
+ long futureEnd = 2051222400000L; // Jan 1, 2035
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH)
+ .setTimeRange(futureStart, futureEnd)
+ .build())
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify empty list returned
+ assertTrue(outputSegments.isEmpty(), "Should return empty list when all records filtered");
+
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testEmptyPartitions()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "empty_partitions_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create data with only abc and xyz values (no other partition values)
+ List<Object[]> testData = new ArrayList<>();
+ testData.addAll(generatePartitionedData(15, "abc", 1597719600000L));
+ testData.addAll(generatePartitionedData(12, "xyz", 1597719600000L));
+
+ File inputDir = new File(TEMP_DIR, "empty_partitions_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure column partitioner
+ PartitionerConfig columnPartitioner = new PartitionerConfig.Builder()
+ .setPartitionerType(PartitionerFactory.PartitionerType.COLUMN_VALUE)
+ .setColumnName("campaign")
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setPartitionerConfigs(List.of(columnPartitioner))
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Verify only 2 output segments (for abc and xyz, no empty partitions)
+ assertEquals(outputSegments.size(), 2, "Should create exactly 2 output segments");
+
+ // Verify total docs
+ int totalDocs = 0;
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ totalDocs += metadata.getTotalDocs();
+ }
+ assertEquals(totalDocs, 27, "Total docs should be 27 (15 + 12)");
+
+ FileUtils.cleanDirectory(workingDir);
+ }
+
+ @Test
+ public void testTwoColumnPartitioningWithoutTime()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "two_column_partitioning_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Create test data with multiple combinations of campaign and clicks values
+ // campaign: "abc", "xyz", "pqr"
+ // clicks: different ranges to test partitioning
+ List<Object[]> testData = new ArrayList<>();
+ long baseTime = 1597719600000L;
+
+ // Add data for abc campaign with 1000 clicks
+ for (int i = 0; i < 20; i++) {
+ testData.add(new Object[]{"abc", 1000, baseTime + i * 1000});
+ }
+ // Add data for abc campaign with 2000 clicks
+ for (int i = 0; i < 15; i++) {
+ testData.add(new Object[]{"abc", 2000, baseTime + i * 1000});
+ }
+ // Add data for xyz campaign with 1000 clicks
+ for (int i = 0; i < 18; i++) {
+ testData.add(new Object[]{"xyz", 1000, baseTime + i * 1000});
+ }
+ // Add data for xyz campaign with 3000 clicks
+ for (int i = 0; i < 12; i++) {
+ testData.add(new Object[]{"xyz", 3000, baseTime + i * 1000});
+ }
+ // Add data for pqr campaign with 2000 clicks
+ for (int i = 0; i < 10; i++) {
+ testData.add(new Object[]{"pqr", 2000, baseTime + i * 1000});
+ }
+ Collections.shuffle(testData, new Random());
+
+ File inputDir = new File(TEMP_DIR, "two_column_partitioning_input");
+ List<File> inputSegments = createInputSegments(inputDir, testData, 1, _schema);
+
+ // Configure two column partitioners - one for campaign, one for clicks
+ PartitionerConfig campaignPartitioner = new PartitionerConfig.Builder()
+ .setPartitionerType(PartitionerFactory.PartitionerType.COLUMN_VALUE)
+ .setColumnName("campaign")
+ .build();
+ PartitionerConfig clicksPartitioner = new PartitionerConfig.Builder()
+ .setPartitionerType(PartitionerFactory.PartitionerType.COLUMN_VALUE)
+ .setColumnName("clicks")
+ .build();
+
+ SegmentProcessorConfig config = new SegmentProcessorConfig.Builder()
+ .setTableConfig(_tableConfig)
+ .setSchema(_schema)
+ .setPartitionerConfigs(List.of(campaignPartitioner, clicksPartitioner))
+ .build();
+
+ List<File> outputSegments = processSegments(inputSegments, config, workingDir);
+
+ // Expected partitions:
+ // abc + 1000-1099 (20 docs)
+ // abc + 2000-2099 (15 docs)
+ // xyz + 1000-1099 (18 docs)
+ // xyz + 3000-3099 (12 docs)
+ // pqr + 2000-2099 (10 docs)
+ // Total: 5 segments
+ assertEquals(outputSegments.size(), 5, "Should create exactly 5 segments for unique combinations");
+
+ // Verify total document count
+ int totalDocs = 0;
+ for (File segmentDir : outputSegments) {
+ SegmentMetadata metadata = new SegmentMetadataImpl(segmentDir);
+ totalDocs += metadata.getTotalDocs();
+
+ // Each segment should have only one campaign value (cardinality = 1)
+ ColumnMetadata campaignMetadata = metadata.getColumnMetadataFor("campaign");
+ assertNotNull(campaignMetadata, "Campaign column metadata should not be null");
+ assertEquals(campaignMetadata.getCardinality(), 1,
+ "Each segment should have exactly 1 campaign value");
+
+ // Each segment should have clicks values from a specific partition
+ ColumnMetadata clicksMetadata = metadata.getColumnMetadataFor("clicks");
+ assertEquals(clicksMetadata.getCardinality(), 1,
+ "Each segment should have exactly 1 clicks value");
+ }
+ assertEquals(totalDocs, 75, "Total docs should be 75 (20+15+18+12+10)");
+
+ FileUtils.cleanDirectory(workingDir);
+ FileUtils.cleanDirectory(inputDir);
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws IOException {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/SegmentProcessorFrameworkTest.java b/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/SegmentProcessorFrameworkTest.java
index 90dd81e72fb..aefcdfcbff2 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/SegmentProcessorFrameworkTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/segment/processing/framework/SegmentProcessorFrameworkTest.java
@@ -19,222 +19,61 @@
package org.apache.pinot.core.segment.processing.framework;
import java.io.File;
-import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.stream.IntStream;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.core.segment.processing.timehandler.TimeHandler;
import org.apache.pinot.core.segment.processing.timehandler.TimeHandlerConfig;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
-import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
-import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
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.SegmentMetadata;
-import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.config.table.ingestion.ComplexTypeConfig;
-import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.FileFormat;
-import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.spi.data.readers.RecordReaderFactory;
import org.apache.pinot.spi.data.readers.RecordReaderFileConfig;
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;
import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.*;
import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
-import static org.testng.Assert.assertThrows;
-import static org.testng.Assert.assertTrue;
/**
- * End-to-end tests for SegmentProcessorFramework
+ * End-to-end tests for SegmentProcessorFramework using RecordReader-based processing.
*/
-public class SegmentProcessorFrameworkTest {
- private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "SegmentProcessorFrameworkTest");
-
- private List<RecordReader> _singleSegment;
- private List<RecordReader> _multipleSegments;
- private List<RecordReader> _multiValueSegments;
- private List<RecordReader> _recordReaderWithComplexType;
-
-
- private TableConfig _tableConfig;
- private TableConfig _tableConfigNullValueEnabled;
- private TableConfig _tableConfigSegmentNameGeneratorEnabled;
- private TableConfig _tableConfigWithFixedSegmentName;
-
- private Schema _schema;
- private Schema _schemaMV;
- private Schema _schemaWithComplexType;
-
- private final List<Object[]> _rawData =
- Arrays.asList(new Object[]{"abc", 1000, 1597719600000L}, new Object[]{null, 2000, 1597773600000L},
- new Object[]{"abc", null, 1597777200000L}, new Object[]{"abc", 4000, 1597795200000L},
- new Object[]{"abc", 3000, 1597802400000L}, new Object[]{null, null, 1597838400000L},
- new Object[]{"xyz", 4000, 1597856400000L}, new Object[]{null, 1000, 1597878000000L},
- new Object[]{"abc", 7000, 1597881600000L}, new Object[]{"xyz", 6000, 1597892400000L});
-
- private final List<Object[]> _rawDataMultiValue =
- Arrays.asList(new Object[]{new String[]{"a", "b"}, 1000, 1597795200000L},
- new Object[]{null, null, 1597795200000L}, new Object[]{null, 1000, 1597795200000L},
- new Object[]{new String[]{"a", "b"}, null, 1597795200000L});
-
- @BeforeClass
- public void setup()
- throws Exception {
- FileUtils.deleteQuietly(TEMP_DIR);
- FileUtils.forceMkdir(TEMP_DIR);
-
- _tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
- _tableConfigNullValueEnabled =
- new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time")
- .setNullHandlingEnabled(true).build();
- _tableConfigSegmentNameGeneratorEnabled =
- new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
- _tableConfigSegmentNameGeneratorEnabled.getIndexingConfig().setSegmentNameGeneratorType("normalizedDate");
- _tableConfigWithFixedSegmentName =
- new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
- _tableConfigWithFixedSegmentName.getIndexingConfig().setSegmentNameGeneratorType("fixed");
-
- _schema =
- new Schema.SchemaBuilder().setSchemaName("mySchema")
- .addSingleValueDimension("campaign", DataType.STRING, "")
- .addSingleValueDimension("campaign.inner1", DataType.STRING)
- .addSingleValueDimension("campaign.inner1.inner2", DataType.STRING)
- // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
- .addMetric("clicks", DataType.INT, 1000)
- .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
- _schemaMV =
- new Schema.SchemaBuilder().setSchemaName("mySchema")
- .addMultiValueDimension("campaign", DataType.STRING, "")
- // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
- .addMetric("clicks", DataType.INT, 1000)
- .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
- _schemaWithComplexType =
- new Schema.SchemaBuilder().setSchemaName("mySchema")
- .addSingleValueDimension("campaign", DataType.JSON)
- .addSingleValueDimension("campaign.inner1", DataType.STRING)
- .addSingleValueDimension("campaign.inner1.inner2", DataType.STRING)
- .addSingleValueDimension("targetusers.user", DataType.STRING)
- // NOTE: Intentionally put 1000 as default value to test skipping null values during rollup
- .addMetric("clicks", DataType.INT, 1000)
- .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
- // create segments in many folders
- _singleSegment = createInputSegments(new File(TEMP_DIR, "single_segment"), _rawData, 1, _schema);
- _multipleSegments = createInputSegments(new File(TEMP_DIR, "multiple_segments"), _rawData, 3, _schema);
- _multiValueSegments =
- createInputSegments(new File(TEMP_DIR, "multi_value_segment"), _rawDataMultiValue, 1, _schemaMV);
- _recordReaderWithComplexType = createRecordReaderWithComplexType();
- }
+public class SegmentProcessorFrameworkTest extends BaseSegmentProcessorFrameworkTest {
- private List<RecordReader> createInputSegments(File inputDir, List<Object[]> rawData, int numSegments, Schema schema)
+ @Override
+ protected List<File> processSegments(List<File> segmentDirs, SegmentProcessorConfig config, File workingDir)
throws Exception {
- assertTrue(inputDir.mkdirs());
-
- List<List<GenericRow>> dataLists = new ArrayList<>();
- if (numSegments > 1) {
- List<GenericRow> dataList1 = new ArrayList<>(4);
- IntStream.range(0, 4).forEach(i -> dataList1.add(getGenericRow(rawData.get(i))));
- dataLists.add(dataList1);
- List<GenericRow> dataList2 = new ArrayList<>(4);
- IntStream.range(4, 7).forEach(i -> dataList2.add(getGenericRow(rawData.get(i))));
- dataLists.add(dataList2);
- List<GenericRow> dataList3 = new ArrayList<>(4);
- IntStream.range(7, 10).forEach(i -> dataList3.add(getGenericRow(rawData.get(i))));
- dataLists.add(dataList3);
- } else {
- List<GenericRow> dataList = new ArrayList<>();
- rawData.forEach(r -> dataList.add(getGenericRow(r)));
- dataLists.add(dataList);
- }
-
- List<RecordReader> segmentRecordReaders = new ArrayList<>(dataLists.size());
- int idx = 0;
- for (List<GenericRow> inputRows : dataLists) {
- RecordReader recordReader = new GenericRowRecordReader(inputRows);
- // NOTE: Generate segments with null value enabled
- SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(_tableConfigNullValueEnabled, schema);
- segmentGeneratorConfig.setOutDir(inputDir.getPath());
- segmentGeneratorConfig.setSequenceId(idx++);
- SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
- driver.init(segmentGeneratorConfig, recordReader);
- driver.build();
- PinotSegmentRecordReader segmentRecordReader = new PinotSegmentRecordReader();
- segmentRecordReader.init(driver.getOutputDirectory(), null, null, true);
- segmentRecordReaders.add(segmentRecordReader);
- }
- return segmentRecordReaders;
- }
-
- private List<RecordReader> createRecordReaderWithComplexType() {
- GenericRow genericRow = new GenericRow();
- genericRow.putValue("a", 1L);
- Map<String, Object> map1 = new HashMap<>();
- genericRow.putValue("campaign", map1);
- map1.put("inner", "innerv");
- Map<String, Object> innerMap1 = new HashMap<>();
- innerMap1.put("inner2", "inner2v");
-
- map1.put("inner1", innerMap1);
- Map<String, Object> map2 = new HashMap<>();
- map2.put("c", 3);
- genericRow.putValue("map2", map2);
-
-
- //list with two map entries inside
- List<Map<String, Object>> list = new ArrayList<>();
- Map<String, Object> map3 = new HashMap<>();
- map3.put("user", "foobar");
- list.add(map3);
- Map<String, Object> map4 = new HashMap<>();
- map4.put("user", "barfoo");
- list.add(map4);
- genericRow.putValue("targetusers", list);
- return List.of(new GenericRowRecordReader(List.of(genericRow)));
- }
-
- private GenericRow getGenericRow(Object[] rawRow) {
- GenericRow row = new GenericRow();
- row.putValue("campaign", rawRow[0]);
- row.putValue("clicks", rawRow[1]);
- row.putValue("time", rawRow[2]);
- return row;
- }
-
- private void rewindRecordReaders(List<RecordReader> recordReaders)
- throws IOException {
- for (RecordReader recordReader : recordReaders) {
- recordReader.rewind();
+ List<RecordReaderFileConfig> recordReaderConfigs = new ArrayList<>();
+ for (File segmentDir : segmentDirs) {
+ PinotSegmentRecordReader reader = new PinotSegmentRecordReader();
+ reader.init(segmentDir, null, null, true);
+ recordReaderConfigs.add(new RecordReaderFileConfig(reader));
}
+ SegmentProcessorFramework framework =
+ new SegmentProcessorFramework(config, workingDir, recordReaderConfigs, Collections.emptyList(), null);
+ return framework.process();
}
/**
* Test lazy initialization of record readers. Here we create
* RecoderReaderFileConfig and the actual reader is initialized during the
* map phase.
- * @throws Exception
*/
@Test
public void testRecordReaderFileConfigInit() throws Exception {
@@ -242,6 +81,7 @@ public class SegmentProcessorFrameworkTest {
FileUtils.forceMkdir(workingDir);
ClassLoader classLoader = getClass().getClassLoader();
URL resource = classLoader.getResource("data/dimBaseballTeams.csv");
+ assertNotNull(resource);
RecordReader recordReader = RecordReaderFactory.getRecordReader(FileFormat.CSV, new File(resource.toURI()),
null, null);
RecordReaderFileConfig recordReaderFileConfig = new RecordReaderFileConfig(FileFormat.CSV,
@@ -252,9 +92,9 @@ public class SegmentProcessorFrameworkTest {
Schema schema =
new Schema.SchemaBuilder().setSchemaName("mySchema").addSingleValueDimension("teamId",
- DataType.STRING, "")
- .addSingleValueDimension("teamName", DataType.STRING, "")
- .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ FieldSpec.DataType.STRING, "")
+ .addSingleValueDimension("teamName", FieldSpec.DataType.STRING, "")
+ .addDateTime("time", FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
SegmentProcessorConfig config =
new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema).build();
@@ -266,404 +106,7 @@ public class SegmentProcessorFrameworkTest {
SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
assertEquals(segmentMetadata.getTotalDocs(), 52);
// Verify reader is closed
- assertEquals(recordReaderFileConfig.isRecordReaderClosedFromRecordReaderFileConfig(), true);
- }
-
- @Test
- public void testSegmentGenerationWithComplexType() throws Exception {
- File workingDir = new File(TEMP_DIR, "single_segment_complex_type_output");
- FileUtils.forceMkdir(workingDir);
- IngestionConfig ingestionConfig = new IngestionConfig();
- List<String> fieldsToUnnest = new ArrayList<>();
- fieldsToUnnest.add("targetusers");
-
- ingestionConfig.setComplexTypeConfig(
- new ComplexTypeConfig(fieldsToUnnest, ".", null, null));
- _tableConfig.setIngestionConfig(ingestionConfig);
- // Default configs
- SegmentProcessorConfig config =
- new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schemaWithComplexType).build();
- SegmentProcessorFramework framework =
- new SegmentProcessorFramework(_recordReaderWithComplexType, config, workingDir);
- List<File> outputSegments = framework.process();
- ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
- SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
- // Pick the column created from complex type
- ColumnMetadata campaignInner2Metadata = segmentMetadata.getColumnMetadataFor("campaign.inner1.inner2");
- // Verify we see a specific value parsed from the complexType
- Assert.assertEquals(campaignInner2Metadata.getMinValue().compareTo("inner2v"), 0);
- ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
- Assert.assertEquals(
- campaignMetadata.getMinValue().compareTo("{\"inner1\":{\"inner2\":\"inner2v\"},\"inner\":\"innerv\"}"), 0);
-
- ColumnMetadata listMetadata = segmentMetadata.getColumnMetadataFor("targetusers.user");
- Assert.assertEquals(listMetadata.getMinValue().compareTo("barfoo"), 0);
- }
-
- @Test
- public void testSingleSegment()
- throws Exception {
- File workingDir = new File(TEMP_DIR, "single_segment_output");
- FileUtils.forceMkdir(workingDir);
-
- // Default configs
- SegmentProcessorConfig config =
- new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
- SegmentProcessorFramework framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- List<File> outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
- SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
- assertEquals(segmentMetadata.getTotalDocs(), 10);
- ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
- assertEquals(campaignMetadata.getCardinality(), 3);
- assertEquals(campaignMetadata.getMinValue(), "");
- assertEquals(campaignMetadata.getMaxValue(), "xyz");
- ColumnMetadata clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
- assertEquals(clicksMetadata.getCardinality(), 6);
- assertEquals(clicksMetadata.getMinValue(), 1000);
- assertEquals(clicksMetadata.getMaxValue(), 7000);
- ColumnMetadata timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 10);
- assertEquals(timeMetadata.getMinValue(), 1597719600000L);
- assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
- DataSource campaignDataSource = segment.getDataSource("campaign");
- assertNull(campaignDataSource.getNullValueVector());
- DataSource clicksDataSource = segment.getDataSource("clicks");
- assertNull(clicksDataSource.getNullValueVector());
- DataSource timeDataSource = segment.getDataSource("time");
- assertNull(timeDataSource.getNullValueVector());
- assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
- segment.destroy();
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Default configs - null value enabled
- config =
- new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigNullValueEnabled).setSchema(_schema).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- String[] outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
- segmentMetadata = segment.getSegmentMetadata();
- assertEquals(segmentMetadata.getTotalDocs(), 10);
- campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
- assertEquals(campaignMetadata.getCardinality(), 3);
- assertEquals(campaignMetadata.getMinValue(), "");
- assertEquals(campaignMetadata.getMaxValue(), "xyz");
- clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
- assertEquals(clicksMetadata.getCardinality(), 6);
- assertEquals(clicksMetadata.getMinValue(), 1000);
- assertEquals(clicksMetadata.getMaxValue(), 7000);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 10);
- assertEquals(timeMetadata.getMinValue(), 1597719600000L);
- assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
- campaignDataSource = segment.getDataSource("campaign");
- NullValueVectorReader campaignNullValueVector = campaignDataSource.getNullValueVector();
- assertNotNull(campaignNullValueVector);
- assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{1, 5, 7});
- clicksDataSource = segment.getDataSource("clicks");
- NullValueVectorReader clicksNullValueVector = clicksDataSource.getNullValueVector();
- assertNotNull(clicksNullValueVector);
- assertEquals(clicksNullValueVector.getNullBitmap().toArray(), new int[]{2, 5});
- timeDataSource = segment.getDataSource("time");
- assertNull(timeDataSource.getNullValueVector());
- assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
- segment.destroy();
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Fixed segment name
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigWithFixedSegmentName).setSchema(_schema)
- .setSegmentConfig(new SegmentConfig.Builder().setFixedSegmentName("myTable_segment_0001").build()).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
- segmentMetadata = segment.getSegmentMetadata();
- assertEquals(segmentMetadata.getName(), "myTable_segment_0001");
- segment.destroy();
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time filter
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597795200000L, 1597881600000L).build())
- .build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 5);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 5);
- assertEquals(timeMetadata.getMinValue(), 1597795200000L);
- assertEquals(timeMetadata.getMaxValue(), 1597878000000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597878000000_0");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Negate time filter
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597795200000L, 1597881600000L)
- .setNegateWindowFilter(true).build()).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 5);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 5);
- assertEquals(timeMetadata.getMinValue(), 1597719600000L);
- assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time filter - filtered everything
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597968000000L, 1598054400000L).build())
- .build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertTrue(outputSegments.isEmpty());
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time round
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema)
- .setTimeHandlerConfig(new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).build())
- .build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 10);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 3);
- assertEquals(timeMetadata.getMinValue(), 1597708800000L);
- assertEquals(timeMetadata.getMaxValue(), 1597881600000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597881600000_0");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time partition
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setPartitionBucketMs(86400000).build()).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 3);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- outputSegments.sort(null);
- // segment 0
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 3);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 3);
- assertEquals(timeMetadata.getMinValue(), 1597719600000L);
- assertEquals(timeMetadata.getMaxValue(), 1597777200000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597777200000_0");
- // segment 1
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
- assertEquals(segmentMetadata.getTotalDocs(), 5);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 5);
- assertEquals(timeMetadata.getMinValue(), 1597795200000L);
- assertEquals(timeMetadata.getMaxValue(), 1597878000000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597878000000_1");
- // segment 2
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 2);
- assertEquals(timeMetadata.getMinValue(), 1597881600000L);
- assertEquals(timeMetadata.getMaxValue(), 1597892400000L);
- assertEquals(segmentMetadata.getName(), "myTable_1597881600000_1597892400000_2");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time filter, round, partition, rollup - null value enabled
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigNullValueEnabled).setSchema(_schema)
- .setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setTimeRange(1597708800000L, 1597881600000L)
- .setRoundBucketMs(86400000).setPartitionBucketMs(86400000).build()).setMergeType(MergeType.ROLLUP)
- .build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 2);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- outputSegments.sort(null);
- // segment 0
- segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
- segmentMetadata = segment.getSegmentMetadata();
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
- assertEquals(campaignMetadata.getCardinality(), 2);
- assertEquals(campaignMetadata.getMinValue(), "");
- assertEquals(campaignMetadata.getMaxValue(), "abc");
- clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
- assertEquals(clicksMetadata.getCardinality(), 2);
- assertEquals(clicksMetadata.getMinValue(), 1000);
- assertEquals(clicksMetadata.getMaxValue(), 2000);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 1);
- assertEquals(timeMetadata.getMinValue(), 1597708800000L);
- assertEquals(timeMetadata.getMaxValue(), 1597708800000L);
- campaignDataSource = segment.getDataSource("campaign");
- campaignNullValueVector = campaignDataSource.getNullValueVector();
- assertNotNull(campaignNullValueVector);
- assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{0});
- clicksDataSource = segment.getDataSource("clicks");
- assertNull(clicksDataSource.getNullValueVector());
- timeDataSource = segment.getDataSource("time");
- assertNull(timeDataSource.getNullValueVector());
- assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597708800000_0");
- segment.destroy();
- // segment 1
- segment = ImmutableSegmentLoader.load(outputSegments.get(1), ReadMode.mmap);
- segmentMetadata = segment.getSegmentMetadata();
- assertEquals(segmentMetadata.getTotalDocs(), 3);
- campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
- assertEquals(campaignMetadata.getCardinality(), 3);
- assertEquals(campaignMetadata.getMinValue(), "");
- assertEquals(campaignMetadata.getMaxValue(), "xyz");
- clicksMetadata = segmentMetadata.getColumnMetadataFor("clicks");
- assertEquals(clicksMetadata.getCardinality(), 3);
- assertEquals(clicksMetadata.getMinValue(), 1000);
- assertEquals(clicksMetadata.getMaxValue(), 7000);
- timeMetadata = segmentMetadata.getColumnMetadataFor("time");
- assertEquals(timeMetadata.getCardinality(), 1);
- assertEquals(timeMetadata.getMinValue(), 1597795200000L);
- assertEquals(timeMetadata.getMaxValue(), 1597795200000L);
- timeDataSource = segment.getDataSource("campaign");
- campaignNullValueVector = timeDataSource.getNullValueVector();
- assertNotNull(campaignNullValueVector);
- assertEquals(campaignNullValueVector.getNullBitmap().toArray(), new int[]{0});
- clicksDataSource = segment.getDataSource("clicks");
- assertNull(clicksDataSource.getNullValueVector());
- timeDataSource = segment.getDataSource("time");
- assertNull(timeDataSource.getNullValueVector());
- assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_1");
- segment.destroy();
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Time round, dedup
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema)
- .setTimeHandlerConfig(new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).build())
- .setMergeType(MergeType.DEDUP).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 8);
- assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597881600000_0");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- // Segment config
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setSegmentConfig(
- new SegmentConfig.Builder().setMaxNumRecordsPerSegment(4).setSegmentNamePrefix("myPrefix")
- .setSegmentNamePostfix("myPostfix").build()).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 3);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- outputSegments.sort(null);
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 4);
- assertEquals(segmentMetadata.getName(), "myPrefix_1597719600000_1597795200000_myPostfix_0");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
- assertEquals(segmentMetadata.getTotalDocs(), 4);
- assertEquals(segmentMetadata.getName(), "myPrefix_1597802400000_1597878000000_myPostfix_1");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- assertEquals(segmentMetadata.getName(), "myPrefix_1597881600000_1597892400000_myPostfix_2");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
-
- config =
- new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigSegmentNameGeneratorEnabled).setSchema(_schema)
- .setSegmentConfig(new SegmentConfig.Builder().setMaxNumRecordsPerSegment(4).setSegmentNamePrefix("myPrefix")
- .setSegmentNamePostfix("myPostfix").build()).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 3);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- outputSegments.sort(null);
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 4);
- assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-18_2020-08-19_myPostfix_0");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
- assertEquals(segmentMetadata.getTotalDocs(), 4);
- assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-19_2020-08-19_myPostfix_1");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- assertEquals(segmentMetadata.getName(), "myPrefix_2020-08-20_2020-08-20_myPostfix_2");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
- }
-
- @Test
- public void testMultipleSegments()
- throws Exception {
- File workingDir = new File(TEMP_DIR, "multiple_segments_output");
- FileUtils.forceMkdir(workingDir);
-
- // Default configs
- SegmentProcessorConfig config =
- new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
- SegmentProcessorFramework framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
- List<File> outputSegments = framework.process();
- assertEquals(outputSegments.size(), 1);
- String[] outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- SegmentMetadata segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 10);
- assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multipleSegments);
-
- // Time round, partition, rollup
- config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
- new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).setPartitionBucketMs(86400000)
- .build()).setMergeType(MergeType.ROLLUP).build();
- framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
- outputSegments = framework.process();
- assertEquals(outputSegments.size(), 3);
- outputDirs = workingDir.list();
- assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
- outputSegments.sort(null);
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597708800000_0");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
- assertEquals(segmentMetadata.getTotalDocs(), 3);
- assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_1");
- segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
- assertEquals(segmentMetadata.getTotalDocs(), 2);
- assertEquals(segmentMetadata.getName(), "myTable_1597881600000_1597881600000_2");
- FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multipleSegments);
+ assertTrue(recordReaderFileConfig.isRecordReaderClosedFromRecordReaderFileConfig());
}
@Test
@@ -673,12 +116,11 @@ public class SegmentProcessorFrameworkTest {
FileUtils.forceMkdir(workingDir);
int expectedTotalDocsCount = 10;
- // Test 1 : Default case i.e. no limit to mapper output file size (single record reader).
+ // Test 1 : Default case i.e. no limit to mapper output file size (single segment).
SegmentProcessorConfig config =
new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
- SegmentProcessorFramework framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- List<File> outputSegments = framework.process();
+ List<File> outputSegments = processSegments(_singleSegment, config, workingDir);
assertEquals(outputSegments.size(), 1);
String[] outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -686,12 +128,10 @@ public class SegmentProcessorFrameworkTest {
assertEquals(segmentMetadata.getTotalDocs(), expectedTotalDocsCount);
assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
- // Test 2 : Default case i.e. no limit to mapper output file size (multiple record readers).
+ // Test 2 : Default case i.e. no limit to mapper output file size (multiple segments).
config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
- framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
- outputSegments = framework.process();
+ outputSegments = processSegments(_multipleSegments, config, workingDir);
assertEquals(outputSegments.size(), 1);
outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -699,9 +139,8 @@ public class SegmentProcessorFrameworkTest {
assertEquals(segmentMetadata.getTotalDocs(), expectedTotalDocsCount);
assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multipleSegments);
- // Test 3 : Test mapper with threshold output size (single record reader).
+ // Test 3 : Test mapper with threshold output size (single segment).
// Create a segmentConfig with intermediate mapper output size threshold set to the number of bytes in each row
// from the data. In this way, we can test if each row is written to a separate segment.
@@ -710,8 +149,7 @@ public class SegmentProcessorFrameworkTest {
.setSegmentNamePostfix("testPostfix").build();
config = new SegmentProcessorConfig.Builder().setSegmentConfig(segmentConfig).setTableConfig(_tableConfig)
.setSchema(_schema).build();
- framework = new SegmentProcessorFramework(_singleSegment, config, workingDir);
- outputSegments = framework.process();
+ outputSegments = processSegments(_singleSegment, config, workingDir);
assertEquals(outputSegments.size(), expectedTotalDocsCount);
outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -724,9 +162,8 @@ public class SegmentProcessorFrameworkTest {
assertTrue(segmentMetadata.getName().matches("testPrefix_.*_testPostfix_" + i));
}
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_singleSegment);
- // Test 4 : Test mapper with threshold output size (multiple record readers).
+ // Test 4 : Test mapper with threshold output size (multiple segments).
// Create a segmentConfig with intermediate mapper output size threshold set to the number of bytes in each row
// from the data. In this way, we can test if each row is written to a separate segment.
@@ -734,8 +171,7 @@ public class SegmentProcessorFrameworkTest {
.setSegmentNamePostfix("testPostfix").build();
config = new SegmentProcessorConfig.Builder().setSegmentConfig(segmentConfig).setTableConfig(_tableConfig)
.setSchema(_schema).build();
- framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
- outputSegments = framework.process();
+ outputSegments = processSegments(_multipleSegments, config, workingDir);
assertEquals(outputSegments.size(), expectedTotalDocsCount);
outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -748,34 +184,47 @@ public class SegmentProcessorFrameworkTest {
assertTrue(segmentMetadata.getName().matches("testPrefix_.*_testPostfix_" + i));
}
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multipleSegments);
// Test 5 : Test with injected failure in mapper to verify output directory is cleaned up.
- List<RecordReader> testList = new ArrayList<>(_multipleSegments);
- testList.set(1, null);
+ List<File> testList = new ArrayList<>(_multipleSegments);
segmentConfig = new SegmentConfig.Builder().setIntermediateFileSizeThreshold(16).setSegmentNamePrefix("testPrefix")
.setSegmentNamePostfix("testPostfix").build();
config = new SegmentProcessorConfig.Builder().setSegmentConfig(segmentConfig).setTableConfig(_tableConfig)
.setSchema(_schema).build();
- SegmentProcessorFramework failureTest = new SegmentProcessorFramework(testList, config, workingDir);
+ processSegments(testList, config, workingDir);
+ List<RecordReader> readers = new ArrayList<>();
+
+ int index = 0;
+ for (File segmentDir : testList) {
+ if (index == 1) {
+ // Inject failure
+ readers.add(null);
+ } else {
+ PinotSegmentRecordReader reader = new PinotSegmentRecordReader();
+ reader.init(segmentDir, null, null, true);
+ readers.add(reader);
+ }
+ index++;
+ }
+ SegmentProcessorFramework failureTest = new SegmentProcessorFramework(readers, config, workingDir);
assertThrows(NullPointerException.class, failureTest::process);
assertTrue(FileUtils.isEmptyDirectory(workingDir));
- rewindRecordReaders(_multipleSegments);
// Test 6: RecordReader should be closed when recordReader is created inside RecordReaderFileConfig (without mapper
// output size threshold configured).
ClassLoader classLoader = getClass().getClassLoader();
URL resource = classLoader.getResource("data/dimBaseballTeams.csv");
+ assertNotNull(resource);
RecordReaderFileConfig recordReaderFileConfig =
new RecordReaderFileConfig(FileFormat.CSV, new File(resource.toURI()), null, null, null);
TableConfig tableConfig =
new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("time").build();
- Schema schema =
- new Schema.SchemaBuilder().setSchemaName("mySchema").addSingleValueDimension("teamId", DataType.STRING, "")
- .addSingleValueDimension("teamName", DataType.STRING, "")
- .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("mySchema")
+ .addSingleValueDimension("teamId", FieldSpec.DataType.STRING, "")
+ .addSingleValueDimension("teamName", FieldSpec.DataType.STRING, "")
+ .addDateTime("time", FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
config = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema).build();
@@ -867,6 +316,45 @@ public class SegmentProcessorFrameworkTest {
FileUtils.cleanDirectory(workingDir);
}
+ @Test
+ public void testMultipleSegments()
+ throws Exception {
+ File workingDir = new File(TEMP_DIR, "multiple_segments_output");
+ FileUtils.forceMkdir(workingDir);
+
+ // Default configs
+ SegmentProcessorConfig config =
+ new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
+ List<File> outputSegments = processSegments(_multipleSegments, config, workingDir);
+ assertEquals(outputSegments.size(), 1);
+ String[] outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 10);
+ assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
+ FileUtils.cleanDirectory(workingDir);
+
+ // Time round, partition, rollup
+ config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
+ new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).setPartitionBucketMs(86400000)
+ .build()).setMergeType(MergeType.ROLLUP).build();
+ outputSegments = processSegments(_multipleSegments, config, workingDir);
+ assertEquals(outputSegments.size(), 3);
+ outputDirs = workingDir.list();
+ assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
+ outputSegments.sort(null);
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597708800000_0");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
+ assertEquals(segmentMetadata.getTotalDocs(), 3);
+ assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_1");
+ segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
+ assertEquals(segmentMetadata.getTotalDocs(), 2);
+ assertEquals(segmentMetadata.getName(), "myTable_1597881600000_1597881600000_2");
+ FileUtils.cleanDirectory(workingDir);
+ }
+
@Test
public void testMultiValue()
throws Exception {
@@ -877,8 +365,7 @@ public class SegmentProcessorFrameworkTest {
SegmentProcessorConfig config =
new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigNullValueEnabled).setSchema(_schemaMV)
.setMergeType(MergeType.ROLLUP).build();
- SegmentProcessorFramework framework = new SegmentProcessorFramework(_multiValueSegments, config, workingDir);
- List<File> outputSegments = framework.process();
+ List<File> outputSegments = processSegments(_multiValueSegments, config, workingDir);
assertEquals(outputSegments.size(), 1);
String[] outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -908,13 +395,11 @@ public class SegmentProcessorFrameworkTest {
assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_0");
segment.destroy();
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multiValueSegments);
// Dedup
config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfigNullValueEnabled).setSchema(_schemaMV)
.setMergeType(MergeType.DEDUP).build();
- framework = new SegmentProcessorFramework(_multiValueSegments, config, workingDir);
- outputSegments = framework.process();
+ outputSegments = processSegments(_multiValueSegments, config, workingDir);
assertEquals(outputSegments.size(), 1);
outputDirs = workingDir.list();
assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
@@ -936,21 +421,5 @@ public class SegmentProcessorFrameworkTest {
assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_0");
segment.destroy();
FileUtils.cleanDirectory(workingDir);
- rewindRecordReaders(_multiValueSegments);
- }
-
- @AfterClass
- public void tearDown()
- throws IOException {
- for (RecordReader recordReader : _singleSegment) {
- recordReader.close();
- }
- for (RecordReader recordReader : _multipleSegments) {
- recordReader.close();
- }
- for (RecordReader recordReader : _multiValueSegments) {
- recordReader.close();
- }
- FileUtils.deleteQuietly(TEMP_DIR);
}
}
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/BaseSegmentCreator.java
similarity index 55%
copy from pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
copy to pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/BaseSegmentCreator.java
index 5e53464ad44..969f2f15a08 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/BaseSegmentCreator.java
@@ -18,51 +18,69 @@
*/
package org.apache.pinot.segment.local.segment.creator.impl;
-import com.fasterxml.jackson.core.JsonParseException;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import java.io.Closeable;
+import java.io.DataOutputStream;
import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
+import java.util.NavigableMap;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.ex.ConfigurationException;
+import org.apache.pinot.common.metrics.MinionMeter;
+import org.apache.pinot.common.metrics.MinionMetrics;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.utils.FileUtils;
import org.apache.pinot.segment.local.io.util.PinotDataBitSet;
import org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator;
+import org.apache.pinot.segment.local.segment.index.converter.SegmentFormatConverterFactory;
import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexPlugin;
import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
-import org.apache.pinot.segment.local.segment.index.forward.ForwardIndexType;
-import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
-import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.segment.index.loader.invertedindex.MultiColumnTextIndexHandler;
+import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder;
+import org.apache.pinot.segment.local.utils.CrcUtils;
import org.apache.pinot.segment.spi.V1Constants;
-import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
+import org.apache.pinot.segment.spi.converter.SegmentFormatConverter;
import org.apache.pinot.segment.spi.creator.ColumnIndexCreationInfo;
+import org.apache.pinot.segment.spi.creator.ColumnStatistics;
import org.apache.pinot.segment.spi.creator.IndexCreationContext;
import org.apache.pinot.segment.spi.creator.SegmentCreator;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.segment.spi.index.IndexCreator;
+import org.apache.pinot.segment.spi.index.IndexHandler;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.TextIndexConfig;
import org.apache.pinot.segment.spi.index.creator.ForwardIndexCreator;
import org.apache.pinot.segment.spi.index.creator.SegmentIndexCreationInfo;
-import org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderContext;
+import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderRegistry;
import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.apache.pinot.spi.config.instance.InstanceType;
import org.apache.pinot.spi.config.table.IndexConfig;
import org.apache.pinot.spi.config.table.SegmentZKPropsConfig;
+import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
@@ -70,9 +88,9 @@ import org.apache.pinot.spi.data.FieldSpec;
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.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.TimeUtils;
import org.joda.time.DateTimeZone;
import org.joda.time.Interval;
@@ -85,169 +103,215 @@ import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Segment.*;
/**
- * Segment creator which writes data in a columnar form.
+ * Abstract base class for segment creators containing common functionality and metadata handling.
*/
-// TODO: check resource leaks
-public class SegmentColumnarIndexCreator implements SegmentCreator {
- // TODO Refactor class name to match interface name
- private static final Logger LOGGER = LoggerFactory.getLogger(SegmentColumnarIndexCreator.class);
+public abstract class BaseSegmentCreator implements SegmentCreator {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseSegmentCreator.class);
// Allow at most 512 characters for the metadata property
- static final int METADATA_PROPERTY_LENGTH_LIMIT = 512;
+ private static final int METADATA_PROPERTY_LENGTH_LIMIT = 512;
+
+ protected Schema _schema;
+ protected Map<String, ColumnIndexCreators> _colIndexes;
+ protected NavigableMap<String, ColumnIndexCreationInfo> _indexCreationInfoMap;
+
+ private int _totalDocs;
private SegmentGeneratorConfig _config;
- private TreeMap<String, ColumnIndexCreationInfo> _indexCreationInfoMap;
- private final Map<String, SegmentDictionaryCreator> _dictionaryCreatorMap = new HashMap<>();
- /**
- * Contains, indexed by column name, the creator associated with each index type.
- *
- * Indexes whose build lifecycle is not DURING_SEGMENT_CREATION are not included here.
- */
- private Map<String, Map<IndexType<?, ?, ?>, IndexCreator>> _creatorsByColAndIndex = new HashMap<>();
- private final Map<String, NullValueVectorCreator> _nullValueVectorCreatorMap = new HashMap<>();
private String _segmentName;
- private Schema _schema;
private File _indexDir;
- private int _totalDocs;
- private int _docIdCounter;
+ @Nullable
+ private InstanceType _instanceType;
- @Override
- public void init(SegmentGeneratorConfig segmentCreationSpec, SegmentIndexCreationInfo segmentIndexCreationInfo,
- TreeMap<String, ColumnIndexCreationInfo> indexCreationInfoMap, Schema schema, File outDir,
- @Nullable int[] immutableToMutableIdMap)
+ /**
+ * Common initialization logic for setting up directory and basic fields.
+ */
+ protected void initializeCommon(SegmentGeneratorConfig segmentCreationSpec, SegmentIndexCreationInfo creationInfo,
+ NavigableMap<String, ColumnIndexCreationInfo> indexCreationInfoMap, Schema schema, File outDir,
+ Map<String, ColumnIndexCreators> colIndexes, @Nullable int[] immutableToMutableIdMap,
+ @Nullable InstanceType instanceType)
throws Exception {
- _docIdCounter = 0;
- _config = segmentCreationSpec;
- _indexCreationInfoMap = indexCreationInfoMap;
-
// Check that the output directory does not exist
Preconditions.checkState(!outDir.exists(), "Segment output directory: %s already exists", outDir);
-
Preconditions.checkState(outDir.mkdirs(), "Failed to create output directory: %s", outDir);
- _indexDir = outDir;
+ _config = segmentCreationSpec;
+ _colIndexes = colIndexes;
+ _indexCreationInfoMap = indexCreationInfoMap;
+ _indexDir = outDir;
_schema = schema;
- _totalDocs = segmentIndexCreationInfo.getTotalDocs();
- if (_totalDocs == 0) {
- return;
- }
+ _totalDocs = creationInfo.getTotalDocs();
+ _instanceType = instanceType;
- Map<String, FieldIndexConfigs> indexConfigs = segmentCreationSpec.getIndexConfigsByColName();
-
- _creatorsByColAndIndex = Maps.newHashMapWithExpectedSize(indexConfigs.keySet().size());
+ initColSegmentCreationInfo(immutableToMutableIdMap);
+ }
+ private void initColSegmentCreationInfo(@Nullable int[] immutableToMutableIdMap)
+ throws Exception {
+ Map<String, FieldIndexConfigs> indexConfigs = _config.getIndexConfigsByColName();
for (String columnName : indexConfigs.keySet()) {
- FieldSpec fieldSpec = schema.getFieldSpecFor(columnName);
- Preconditions.checkState(fieldSpec != null, "Failed to find column: %s in the schema", columnName);
- if (fieldSpec.isVirtualColumn()) {
- LOGGER.warn("Ignoring index creation for virtual column {}", columnName);
- continue;
+ if (canColumnBeIndexed(columnName) && _totalDocs > 0 && _indexCreationInfoMap.containsKey(columnName)) {
+ ColumnIndexCreators result = createColIndexCreators(columnName, immutableToMutableIdMap);
+ _colIndexes.put(columnName, result);
+ } else {
+ FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
+ _colIndexes.put(columnName,
+ new ColumnIndexCreators(columnName, fieldSpec, null, List.of(), null));
}
+ }
+ }
- FieldIndexConfigs originalConfig = indexConfigs.get(columnName);
- ColumnIndexCreationInfo columnIndexCreationInfo = indexCreationInfoMap.get(columnName);
- Preconditions.checkNotNull(columnIndexCreationInfo, "Missing index creation info for column: %s", columnName);
- boolean dictEnabledColumn = createDictionaryForColumn(columnIndexCreationInfo, segmentCreationSpec, fieldSpec);
- if (originalConfig.getConfig(StandardIndexes.inverted()).isEnabled()) {
- Preconditions.checkState(dictEnabledColumn,
- "Cannot create inverted index for raw index column: %s", columnName);
- }
+ /**
+ * Initializes a single column's dictionary and index creators.
+ * This encapsulates the common logic shared between different segment creator implementations.
+ */
+ protected ColumnIndexCreators createColIndexCreators(String columnName,
+ @Nullable int[] immutableToMutableIdMap)
+ throws Exception {
+ FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- IndexType<ForwardIndexConfig, ?, ForwardIndexCreator> forwardIdx = StandardIndexes.forward();
- boolean forwardIndexDisabled = !originalConfig.getConfig(forwardIdx).isEnabled();
-
- //@formatter:off
- IndexCreationContext.Common context = IndexCreationContext.builder()
- .withIndexDir(_indexDir)
- .withDictionary(dictEnabledColumn)
- .withFieldSpec(fieldSpec)
- .withTotalDocs(segmentIndexCreationInfo.getTotalDocs())
- .withColumnIndexCreationInfo(columnIndexCreationInfo)
- .withOptimizedDictionary(_config.isOptimizeDictionary()
- || _config.isOptimizeDictionaryForMetrics() && fieldSpec.getFieldType() == FieldSpec.FieldType.METRIC)
- .onHeap(segmentCreationSpec.isOnHeap())
- .withForwardIndexDisabled(forwardIndexDisabled)
- .withTextCommitOnClose(true)
- .withImmutableToMutableIdMap(immutableToMutableIdMap)
- .withRealtimeConversion(segmentCreationSpec.isRealtimeConversion())
- .withConsumerDir(segmentCreationSpec.getConsumerDir())
- .withTableNameWithType(_config.getTableConfig().getTableName())
- .withContinueOnError(_config.isContinueOnError())
- .build();
- //@formatter:on
-
- FieldIndexConfigs config = adaptConfig(columnName, originalConfig, columnIndexCreationInfo, segmentCreationSpec);
-
- if (dictEnabledColumn) {
- // Create dictionary-encoded index
- // Initialize dictionary creator
- // TODO: Dictionary creator holds all unique values on heap. Consider keeping dictionary instead of creator
- // which uses off-heap memory.
-
- DictionaryIndexConfig dictConfig = config.getConfig(StandardIndexes.dictionary());
- if (!dictConfig.isEnabled()) {
- LOGGER.info("Creating dictionary index in column {}.{} even when it is disabled in config",
- segmentCreationSpec.getTableName(), columnName);
- }
+ FieldIndexConfigs originalConfig = _config.getIndexConfigsByColName().get(columnName);
+ ColumnIndexCreationInfo columnIndexCreationInfo = _indexCreationInfoMap.get(columnName);
+ Preconditions.checkNotNull(columnIndexCreationInfo, "Missing index creation info for column: %s", columnName);
- // override dictionary type if configured to do so
- if (_config.isOptimizeDictionaryType()) {
- LOGGER.info("Overriding dictionary type for column: {} using var-length dictionary: {}", columnName,
- columnIndexCreationInfo.isUseVarLengthDictionary());
- dictConfig = new DictionaryIndexConfig(dictConfig, columnIndexCreationInfo.isUseVarLengthDictionary());
- }
+ boolean dictEnabledColumn = createDictionaryForColumn(columnIndexCreationInfo, _config, fieldSpec);
+ if (originalConfig.getConfig(StandardIndexes.inverted()).isEnabled()) {
+ Preconditions.checkState(dictEnabledColumn,
+ "Cannot create inverted index for raw index column: %s", columnName);
+ }
+ IndexCreationContext.Common context =
+ getIndexCreationContext(fieldSpec, dictEnabledColumn, immutableToMutableIdMap);
- SegmentDictionaryCreator creator =
- new DictionaryIndexPlugin().getIndexType().createIndexCreator(context, dictConfig);
+ FieldIndexConfigs config = adaptConfig(columnName, originalConfig, columnIndexCreationInfo, _config);
- try {
- creator.build(context.getSortedUniqueElementsArray());
- } catch (Exception e) {
- LOGGER.error("Error building dictionary for field: {}, cardinality: {}, number of bytes per entry: {}",
- context.getFieldSpec().getName(), context.getCardinality(), creator.getNumBytesPerEntry());
- throw e;
- }
+ SegmentDictionaryCreator dictionaryCreator = null;
+ if (dictEnabledColumn) {
+ dictionaryCreator = getDictionaryCreator(columnName, originalConfig, context);
+ }
- _dictionaryCreatorMap.put(columnName, creator);
- }
+ List<IndexCreator> indexCreators = getIndexCreatorsByColumn(fieldSpec, context, config, dictEnabledColumn);
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex =
- Maps.newHashMapWithExpectedSize(IndexService.getInstance().getAllIndexes().size());
- for (IndexType<?, ?, ?> index : IndexService.getInstance().getAllIndexes()) {
- if (index.getIndexBuildLifecycle() != IndexType.BuildLifecycle.DURING_SEGMENT_CREATION) {
- continue;
- }
- tryCreateIndexCreator(creatorsByIndex, index, context, config);
+ return new ColumnIndexCreators(columnName, fieldSpec, dictionaryCreator,
+ indexCreators, getNullValueCreator(fieldSpec));
+ }
+
+ private IndexCreationContext.Common getIndexCreationContext(FieldSpec fieldSpec, boolean dictEnabledColumn,
+ @Nullable int[] immutableToMutableIdMap) {
+ ColumnIndexCreationInfo columnIndexCreationInfo = _indexCreationInfoMap.get(fieldSpec.getName());
+ FieldIndexConfigs fieldIndexConfig = _config.getIndexConfigsByColName().get(fieldSpec.getName());
+ boolean forwardIndexDisabled = !fieldIndexConfig.getConfig(StandardIndexes.forward()).isEnabled();
+
+ return IndexCreationContext.builder()
+ .withIndexDir(_indexDir)
+ .withDictionary(dictEnabledColumn)
+ .withFieldSpec(fieldSpec)
+ .withTotalDocs(_totalDocs)
+ .withColumnIndexCreationInfo(columnIndexCreationInfo)
+ .withOptimizedDictionary(_config.isOptimizeDictionary()
+ || _config.isOptimizeDictionaryForMetrics() && fieldSpec.getFieldType() == FieldSpec.FieldType.METRIC)
+ .onHeap(_config.isOnHeap())
+ .withForwardIndexDisabled(forwardIndexDisabled)
+ .withTextCommitOnClose(true)
+ .withImmutableToMutableIdMap(immutableToMutableIdMap)
+ .withRealtimeConversion(_config.isRealtimeConversion())
+ .withConsumerDir(_config.getConsumerDir())
+ .withTableNameWithType(_config.getTableConfig().getTableName())
+ .withContinueOnError(_config.isContinueOnError())
+ .build();
+ }
+
+ private SegmentDictionaryCreator getDictionaryCreator(String columnName, FieldIndexConfigs config,
+ IndexCreationContext.Common context)
+ throws IOException {
+ ColumnIndexCreationInfo columnIndexCreationInfo = _indexCreationInfoMap.get(columnName);
+
+ // Create dictionary-encoded index
+ // Initialize dictionary creator
+ // TODO: Dictionary creator holds all unique values on heap. Consider keeping dictionary instead of creator
+ // which uses off-heap memory.
+ DictionaryIndexConfig dictConfig = config.getConfig(StandardIndexes.dictionary());
+ if (!dictConfig.isEnabled()) {
+ LOGGER.info("Creating dictionary index in column {}.{} even when it is disabled in config",
+ _config.getTableName(), columnName);
+ }
+
+ // override dictionary type if configured to do so
+ if (_config.isOptimizeDictionaryType()) {
+ LOGGER.info("Overriding dictionary type for column: {} using var-length dictionary: {}", columnName,
+ columnIndexCreationInfo.isUseVarLengthDictionary());
+ dictConfig = new DictionaryIndexConfig(dictConfig, columnIndexCreationInfo.isUseVarLengthDictionary());
+ }
+
+ SegmentDictionaryCreator dictionaryCreator =
+ new DictionaryIndexPlugin().getIndexType().createIndexCreator(context, dictConfig);
+
+ try {
+ dictionaryCreator.build(context.getSortedUniqueElementsArray());
+ } catch (Exception e) {
+ LOGGER.error("Error building dictionary for field: {}, cardinality: {}, number of bytes per entry: {}",
+ context.getFieldSpec().getName(), context.getCardinality(), dictionaryCreator.getNumBytesPerEntry());
+ throw e;
+ }
+ return dictionaryCreator;
+ }
+
+ private List<IndexCreator> getIndexCreatorsByColumn(FieldSpec fieldSpec, IndexCreationContext.Common context,
+ FieldIndexConfigs config, boolean dictEnabledColumn)
+ throws Exception {
+ Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex =
+ Maps.newHashMapWithExpectedSize(IndexService.getInstance().getAllIndexes().size());
+ for (IndexType<?, ?, ?> index : IndexService.getInstance().getAllIndexes()) {
+ if (index.getIndexBuildLifecycle() != IndexType.BuildLifecycle.DURING_SEGMENT_CREATION) {
+ continue;
}
- // TODO: Remove this when values stored as ForwardIndex stop depending on TextIndex config
- IndexCreator oldFwdCreator = creatorsByIndex.get(forwardIdx);
- if (oldFwdCreator != null) {
- Object fakeForwardValue = calculateRawValueForTextIndex(dictEnabledColumn, config, fieldSpec);
- if (fakeForwardValue != null) {
- ForwardIndexCreator castedOldFwdCreator = (ForwardIndexCreator) oldFwdCreator;
- SameValueForwardIndexCreator fakeValueFwdCreator =
- new SameValueForwardIndexCreator(fakeForwardValue, castedOldFwdCreator);
- creatorsByIndex.put(forwardIdx, fakeValueFwdCreator);
- }
+ tryCreateIndexCreator(creatorsByIndex, index, context, config);
+ }
+
+ // TODO: Remove this when values stored as ForwardIndex stop depending on TextIndex config
+ IndexCreator oldFwdCreator = creatorsByIndex.get(StandardIndexes.forward());
+ if (oldFwdCreator != null) {
+ Object fakeForwardValue = calculateRawValueForTextIndex(dictEnabledColumn, config, fieldSpec);
+ if (fakeForwardValue != null) {
+ ForwardIndexCreator castedOldFwdCreator = (ForwardIndexCreator) oldFwdCreator;
+ SameValueForwardIndexCreator fakeValueFwdCreator =
+ new SameValueForwardIndexCreator(fakeForwardValue, castedOldFwdCreator);
+ creatorsByIndex.put(StandardIndexes.forward(), fakeValueFwdCreator);
}
- _creatorsByColAndIndex.put(columnName, creatorsByIndex);
}
+ return new ArrayList<>(creatorsByIndex.values());
+ }
+ private NullValueVectorCreator getNullValueCreator(FieldSpec fieldSpec) {
// Although NullValueVector is implemented as an index, it needs to be treated in a different way than other indexes
- for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
- String columnName = fieldSpec.getName();
- if (isNullable(fieldSpec)) {
- // Initialize Null value vector map
- LOGGER.info("Column: {} is nullable", columnName);
- _nullValueVectorCreatorMap.put(columnName, new NullValueVectorCreator(_indexDir, columnName));
- } else {
- LOGGER.info("Column: {} is not nullable", columnName);
- }
+ String columnName = fieldSpec.getName();
+ if (isNullable(fieldSpec)) {
+ // Initialize Null value vector map
+ LOGGER.info("Column: {} is nullable", columnName);
+ return new NullValueVectorCreator(_indexDir, columnName);
+ } else {
+ LOGGER.info("Column: {} is not nullable", columnName);
+ return null;
+ }
+ }
+
+ protected boolean canColumnBeIndexed(String columnName) {
+ FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
+ Preconditions.checkState(fieldSpec != null, "Failed to find column: %s in the schema", columnName);
+ if (fieldSpec.isVirtualColumn()) {
+ LOGGER.warn("Ignoring index creation for virtual column {}", columnName);
+ return false;
}
+ return true;
}
+ /**
+ * Checks if a field is nullable based on schema and config settings.
+ */
private boolean isNullable(FieldSpec fieldSpec) {
return _schema.isEnableColumnBasedNullHandling() ? fieldSpec.isNullable() : _config.isDefaultNullHandlingEnabled();
}
+ /**
+ * Adapts field index configs based on column properties.
+ */
private FieldIndexConfigs adaptConfig(String columnName, FieldIndexConfigs config,
ColumnIndexCreationInfo columnIndexCreationInfo, SegmentGeneratorConfig segmentCreationSpec) {
FieldIndexConfigs.Builder builder = new FieldIndexConfigs.Builder(config);
@@ -314,198 +378,8 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
}
/**
- * @deprecated use {@link ForwardIndexType#getDefaultCompressionType(FieldType)} instead
+ * Calculates the raw value to be used for text index when forward index is disabled.
*/
- @Deprecated
- public static ChunkCompressionType getDefaultCompressionType(FieldType fieldType) {
- return ForwardIndexType.getDefaultCompressionType(fieldType);
- }
-
- @Override
- public void indexRow(GenericRow row)
- throws IOException {
- for (Map.Entry<String, Map<IndexType<?, ?, ?>, IndexCreator>> byColEntry : _creatorsByColAndIndex.entrySet()) {
- String columnName = byColEntry.getKey();
-
- Object columnValueToIndex = row.getValue(columnName);
- if (columnValueToIndex == null) {
- throw new RuntimeException("Null value for column:" + columnName);
- }
-
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex = byColEntry.getValue();
-
- FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
- try {
- if (fieldSpec.isSingleValueField()) {
- indexSingleValueRow(dictionaryCreator, columnValueToIndex, creatorsByIndex);
- } else {
- indexMultiValueRow(dictionaryCreator, (Object[]) columnValueToIndex, creatorsByIndex);
- }
- } catch (JsonParseException jpe) {
- throw new ColumnJsonParserException(columnName, jpe);
- }
- }
-
- for (Map.Entry<String, NullValueVectorCreator> entry : _nullValueVectorCreatorMap.entrySet()) {
- // If row has null value for given column name, add to null value vector
- if (row.isNullValue(entry.getKey())) {
- entry.getValue().setNull(_docIdCounter);
- }
- }
-
- _docIdCounter++;
- }
-
- /**
- * Indexes a column from the given segment.
- *
- * @param columnName The name of the column to index
- * @param sortedDocIds If not null, provides the sorted order of documents for processing
- * @param segment The segment containing the column data
- */
- @Override
- public void indexColumn(String columnName, @Nullable int[] sortedDocIds, IndexSegment segment)
- throws IOException {
- indexColumn(columnName, sortedDocIds, segment, null);
- }
-
- /**
- * Indexes a column from the given segment.
- *
- * @param columnName The name of the column to index
- * @param sortedDocIds If not null, provides the sorted order of documents for processing
- * @param segment The segment containing the column data
- * @param validDocIds If not null, only processes documents that are marked as valid in this bitmap.
- * When null, all documents in the segment are processed. This is used for
- * commit-time compaction to skip invalid/deleted documents during indexing.
- */
- @Override
- public void indexColumn(String columnName, @Nullable int[] sortedDocIds, IndexSegment segment,
- @Nullable ThreadSafeMutableRoaringBitmap validDocIds)
- throws IOException {
- // Iterate over each value in the column
- int numDocs = segment.getSegmentMetadata().getTotalDocs();
- if (numDocs == 0) {
- return;
- }
- try (PinotSegmentColumnReader colReader = new PinotSegmentColumnReader(segment, columnName)) {
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex = _creatorsByColAndIndex.get(columnName);
- NullValueVectorCreator nullVec = _nullValueVectorCreatorMap.get(columnName);
- FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
- if (sortedDocIds != null) {
- int onDiskDocId = 0;
- for (int docId : sortedDocIds) {
- // If validDodIds are provided, only index column if it's a valid doc
- if (validDocIds == null || validDocIds.contains(docId)) {
- indexColumnValue(colReader, creatorsByIndex, columnName, fieldSpec, dictionaryCreator, docId, onDiskDocId,
- nullVec);
- onDiskDocId++;
- }
- }
- } else {
- int onDiskDocId = 0;
- for (int docId = 0; docId < numDocs; docId++) {
- // If validDodIds are provided, only index column if it's a valid doc
- if (validDocIds == null || validDocIds.contains(docId)) {
- indexColumnValue(colReader, creatorsByIndex, columnName, fieldSpec, dictionaryCreator, docId, onDiskDocId,
- nullVec);
- onDiskDocId++;
- }
- }
- }
- }
- }
-
- /**
- * 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<IndexType<?, ?, ?>, 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<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex, String columnName, FieldSpec fieldSpec,
- SegmentDictionaryCreator dictionaryCreator, int sourceDocId, int onDiskDocPos,
- @Nullable NullValueVectorCreator nullVec)
- throws IOException {
- Object columnValueToIndex = colReader.getValue(sourceDocId);
- if (columnValueToIndex == null) {
- throw new RuntimeException("Null value for column:" + columnName);
- }
-
- if (fieldSpec.isSingleValueField()) {
- indexSingleValueRow(dictionaryCreator, columnValueToIndex, creatorsByIndex);
- } else {
- indexMultiValueRow(dictionaryCreator, (Object[]) columnValueToIndex, creatorsByIndex);
- }
-
- if (nullVec != null) {
- if (colReader.isNull(sourceDocId)) {
- nullVec.setNull(onDiskDocPos);
- }
- }
- }
-
- private void indexSingleValueRow(SegmentDictionaryCreator dictionaryCreator, Object value,
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex)
- throws IOException {
- int dictId = dictionaryCreator != null ? dictionaryCreator.indexOfSV(value) : -1;
- for (IndexCreator creator : creatorsByIndex.values()) {
- creator.add(value, dictId);
- }
- }
-
- private void indexMultiValueRow(SegmentDictionaryCreator dictionaryCreator, Object[] values,
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex)
- throws IOException {
- int[] dictId = dictionaryCreator != null ? dictionaryCreator.indexOfMV(values) : null;
- for (IndexCreator creator : creatorsByIndex.values()) {
- creator.add(values, dictId);
- }
- }
-
@Nullable
private Object calculateRawValueForTextIndex(boolean dictEnabledColumn, FieldIndexConfigs configs,
FieldSpec fieldSpec) {
@@ -537,29 +411,10 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
return rawValue;
}
- @Override
- public void setSegmentName(String segmentName) {
- _segmentName = segmentName;
- }
-
- @Override
- public void seal()
- throws ConfigurationException, IOException {
- for (SegmentDictionaryCreator creator : _dictionaryCreatorMap.values()) {
- creator.seal();
- }
- for (NullValueVectorCreator creator : _nullValueVectorCreatorMap.values()) {
- creator.seal();
- }
- for (Map<IndexType<?, ?, ?>, IndexCreator> creatorsByType : _creatorsByColAndIndex.values()) {
- for (IndexCreator creator : creatorsByType.values()) {
- creator.seal();
- }
- }
- writeMetadata();
- }
-
- private void writeMetadata()
+ /**
+ * Writes segment metadata to disk.
+ */
+ protected void writeMetadata()
throws ConfigurationException {
File metadataFile = new File(_indexDir, V1Constants.MetadataKeys.METADATA_FILE_NAME);
PropertiesConfiguration properties = CommonsConfigurationUtils.fromFile(metadataFile);
@@ -680,7 +535,7 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
for (Map.Entry<String, ColumnIndexCreationInfo> entry : _indexCreationInfoMap.entrySet()) {
String column = entry.getKey();
ColumnIndexCreationInfo columnIndexCreationInfo = entry.getValue();
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(column);
+ SegmentDictionaryCreator dictionaryCreator = _colIndexes.get(column).getDictionaryCreator();
int dictionaryElementSize = (dictionaryCreator != null) ? dictionaryCreator.getNumBytesPerEntry() : 0;
addColumnMetadataInfo(properties, column, columnIndexCreationInfo, _totalDocs, _schema.getFieldSpecFor(column),
dictionaryCreator != null, dictionaryElementSize);
@@ -691,10 +546,12 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
properties.setProperty(Realtime.START_OFFSET, segmentZKPropsConfig.getStartOffset());
properties.setProperty(Realtime.END_OFFSET, segmentZKPropsConfig.getEndOffset());
}
-
CommonsConfigurationUtils.saveToFile(properties, metadataFile);
}
+ /**
+ * Adds column metadata information to the properties configuration.
+ */
public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column,
ColumnIndexCreationInfo columnIndexCreationInfo, int totalDocs, FieldSpec fieldSpec, boolean hasDictionary,
int dictionaryElementSize) {
@@ -703,9 +560,9 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
properties.setProperty(getKeyFor(column, TOTAL_DOCS), String.valueOf(totalDocs));
DataType dataType = fieldSpec.getDataType();
properties.setProperty(getKeyFor(column, DATA_TYPE), String.valueOf(dataType));
- // TODO: When the column is raw (no dictionary), we should set BITS_PER_ELEMENT to -1 (invalid). Currently we set it
- // regardless of whether dictionary is created or not for backward compatibility because ForwardIndexHandler
- // doesn't update this value when converting a raw column to dictionary encoded.
+ // TODO: When the column is raw (no dictionary), we should set BITS_PER_ELEMENT to -1 (invalid). Currently we set
+ // it regardless of whether dictionary is created or not for backward compatibility because
+ // ForwardIndexHandler doesn't update this value when converting a raw column to dictionary encoded.
// Consider changing it after releasing 1.5.0.
// See https://github.com/apache/pinot/pull/16921 for details
properties.setProperty(getKeyFor(column, BITS_PER_ELEMENT),
@@ -734,7 +591,8 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
}
}
- PartitionFunction partitionFunction = columnIndexCreationInfo.getPartitionFunction();
+ PartitionFunction partitionFunction =
+ columnIndexCreationInfo.getPartitionFunction();
if (partitionFunction != null) {
properties.setProperty(getKeyFor(column, PARTITION_FUNCTION), partitionFunction.getName());
properties.setProperty(getKeyFor(column, NUM_PARTITIONS), columnIndexCreationInfo.getNumPartitions());
@@ -838,6 +696,9 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
}
}
+ /**
+ * Adds column min/max value information to the properties configuration.
+ */
public static void addColumnMinMaxValueInfo(PropertiesConfiguration properties, String column,
@Nullable Object minValue, @Nullable Object maxValue, DataType storedType) {
String validMinValue = minValue != null ? getValidPropertyValue(minValue.toString(), storedType) : null;
@@ -867,17 +728,259 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
: value;
}
+ /**
+ * Removes column metadata information from the properties configuration.
+ */
public static void removeColumnMetadataInfo(PropertiesConfiguration properties, String column) {
properties.subset(COLUMN_PROPS_KEY_PREFIX + column).clear();
}
+ @Override
+ public String getSegmentName() {
+ return _segmentName;
+ }
+
+ @Override
+ public void seal()
+ throws Exception {
+ _segmentName = generateSegmentName();
+ try {
+ // Write the index files to disk
+ flushColIndexes();
+ } finally {
+ closeColIndexes();
+ }
+ LOGGER.info("Finished segment seal for: {}", _segmentName);
+
+ // Format conversion
+ convertFormatIfNecessary(_indexDir);
+
+ // Build indexes if there are documents
+ if (_totalDocs > 0) {
+ buildStarTreeV2IfNecessary(_indexDir);
+ buildMultiColumnTextIndex(_indexDir);
+ }
+
+ // Update post-creation indexes
+ updatePostSegmentCreationIndexes(_indexDir);
+
+ // Persist creation metadata
+ persistCreationMeta(_indexDir);
+
+ LOGGER.info("Successfully created segment: {} in {}", _segmentName, _indexDir);
+ }
+
+ /**
+ * Generate segment name based on configuration and statistics.
+ * @return Generated segment name
+ */
+ private String generateSegmentName() {
+ ColumnStatistics timeStats = null;
+ String timeColumn = _config.getTimeColumnName();
+ if (timeColumn != null) {
+ timeStats = _indexCreationInfoMap.get(timeColumn).getColumnStatistics();
+ }
+
+ if (timeStats != null) {
+ if (_totalDocs > 0) {
+ return _config.getSegmentNameGenerator()
+ .generateSegmentName(_config.getSequenceId(), timeStats.getMinValue(), timeStats.getMaxValue());
+ } else {
+ // When totalDoc is 0, check whether 'failOnEmptySegment' option is true
+ Preconditions.checkArgument(!_config.isFailOnEmptySegment(),
+ "Failing the empty segment creation as the option 'failOnEmptySegment' is set to: "
+ + _config.isFailOnEmptySegment());
+ // Generate a unique name for a segment with no rows
+ long now = System.currentTimeMillis();
+ return _config.getSegmentNameGenerator().generateSegmentName(_config.getSequenceId(), now, now);
+ }
+ } else {
+ return _config.getSegmentNameGenerator().generateSegmentName(_config.getSequenceId(), null, null);
+ }
+ }
+
+ /**
+ * Writes the individual column index files to disk.
+ */
+ private void flushColIndexes() throws Exception {
+ for (ColumnIndexCreators colIndexes : _colIndexes.values()) {
+ colIndexes.seal();
+ }
+ writeMetadata();
+ }
+
+ private void closeColIndexes() throws IOException {
+ List<Closeable> creators = new ArrayList<>(_colIndexes.values());
+ FileUtils.close(creators);
+ }
+
+ // Explanation of why we are using format converter:
+ // There are 3 options to correctly generate segments to v3 format
+ // 1. Generate v3 directly: This is efficient but v3 index writer needs to know buffer size upfront.
+ // Inverted, star and raw indexes don't have the index size upfront. This is also the least flexible approach
+ // if we add more indexes in the future.
+ // 2. Hold data in-memory: One way to work around predeclaring sizes in (1) is to allocate "large" buffer (2GB?)
+ // and hold the data in memory and write the buffer at the end. The memory requirement in this case increases linearly
+ // with the number of columns. Variation of that is to mmap data to separate files...which is what we are doing here
+ // 3. Another option is to generate dictionary and fwd indexes in v3 and generate inverted, star and raw indexes in
+ // separate files. Then add those files to v3 index file. This leads to lot of hodgepodge code to
+ // handle multiple segment formats.
+ // Using converter is similar to option (2), plus it's battle-tested code. We will roll out with
+ // this change to keep changes limited. Once we've migrated we can implement approach (1) with option to
+ // copy for indexes for which we don't know sizes upfront.
+ private void convertFormatIfNecessary(File segmentDirectory)
+ throws Exception {
+ SegmentVersion versionToGenerate = _config.getSegmentVersion();
+ if (versionToGenerate.equals(SegmentVersion.v1)) {
+ // v1 by default
+ return;
+ }
+ SegmentFormatConverter converter =
+ SegmentFormatConverterFactory.getConverter(SegmentVersion.v1, SegmentVersion.v3);
+ converter.convert(segmentDirectory);
+ }
+
+ /**
+ * Build star-tree V2 index if configured.
+ */
+ private void buildStarTreeV2IfNecessary(File indexDir)
+ throws Exception {
+ List<StarTreeIndexConfig> starTreeIndexConfigs = _config.getStarTreeIndexConfigs();
+ boolean enableDefaultStarTree = _config.isEnableDefaultStarTree();
+ if (CollectionUtils.isNotEmpty(starTreeIndexConfigs) || enableDefaultStarTree) {
+ MultipleTreesBuilder.BuildMode buildMode =
+ _config.isOnHeap() ? MultipleTreesBuilder.BuildMode.ON_HEAP : MultipleTreesBuilder.BuildMode.OFF_HEAP;
+ MultipleTreesBuilder builder = new MultipleTreesBuilder(starTreeIndexConfigs, enableDefaultStarTree, indexDir,
+ buildMode);
+ // We don't create the builder using the try-with-resources pattern because builder.close() performs
+ // some clean-up steps to roll back the star-tree index to the previous state if it exists. If this goes wrong
+ // the star-tree index can be in an inconsistent state. To prevent that, when builder.close() throws an
+ // exception we want to propagate that up instead of ignoring it. This can get clunky when using
+ // try-with-resources as in this scenario the close() exception will be added to the suppressed exception list
+ // rather than thrown as the main exception, even though the original exception thrown on build() is ignored.
+ try {
+ builder.build();
+ } catch (Exception e) {
+ String tableNameWithType = _config.getTableConfig().getTableName();
+ LOGGER.error("Failed to build star-tree index for table: {}, skipping", tableNameWithType, e);
+ // Track metrics only if instance type is provided
+ if (_instanceType != null) {
+ if (_instanceType == InstanceType.MINION) {
+ MinionMetrics.get().addMeteredTableValue(tableNameWithType, MinionMeter.STAR_TREE_INDEX_BUILD_FAILURES, 1);
+ } else {
+ ServerMetrics.get().addMeteredTableValue(tableNameWithType, ServerMeter.STAR_TREE_INDEX_BUILD_FAILURES, 1);
+ }
+ }
+ } finally {
+ builder.close();
+ }
+ }
+ }
+
+ /**
+ * Build multi-column text index if configured.
+ */
+ private void buildMultiColumnTextIndex(File segmentOutputDir)
+ throws Exception {
+ if (_config.getMultiColumnTextIndexConfig() != null) {
+ PinotConfiguration segmentDirectoryConfigs =
+ new PinotConfiguration(Map.of(IndexLoadingConfig.READ_MODE_KEY, ReadMode.mmap));
+
+ TableConfig tableConfig = _config.getTableConfig();
+ Schema schema = _config.getSchema();
+ SegmentDirectoryLoaderContext segmentLoaderContext =
+ new SegmentDirectoryLoaderContext.Builder()
+ .setTableConfig(tableConfig)
+ .setSchema(schema)
+ .setSegmentName(_segmentName)
+ .setSegmentDirectoryConfigs(segmentDirectoryConfigs)
+ .build();
+
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig, schema);
+
+ try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
+ .load(segmentOutputDir.toURI(), segmentLoaderContext);
+ SegmentDirectory.Writer segmentWriter = segmentDirectory.createWriter()) {
+ MultiColumnTextIndexHandler handler = new MultiColumnTextIndexHandler(segmentDirectory, indexLoadingConfig,
+ _config.getMultiColumnTextIndexConfig());
+ handler.updateIndices(segmentWriter);
+ handler.postUpdateIndicesCleanup(segmentWriter);
+ }
+ }
+ }
+
+ /**
+ * Update indexes that are created post-segment creation.
+ */
+ private void updatePostSegmentCreationIndexes(File indexDir)
+ throws Exception {
+ Set<IndexType> postSegCreationIndexes = IndexService.getInstance().getAllIndexes().stream()
+ .filter(indexType -> indexType.getIndexBuildLifecycle() == IndexType.BuildLifecycle.POST_SEGMENT_CREATION)
+ .collect(Collectors.toSet());
+
+ if (!postSegCreationIndexes.isEmpty()) {
+ // Build other indexes
+ Map<String, Object> props = new HashMap<>();
+ props.put(IndexLoadingConfig.READ_MODE_KEY, ReadMode.mmap);
+ PinotConfiguration segmentDirectoryConfigs = new PinotConfiguration(props);
+
+ TableConfig tableConfig = _config.getTableConfig();
+ Schema schema = _config.getSchema();
+ SegmentDirectoryLoaderContext segmentLoaderContext =
+ new SegmentDirectoryLoaderContext.Builder()
+ .setTableConfig(tableConfig)
+ .setSchema(schema)
+ .setSegmentName(_segmentName)
+ .setSegmentDirectoryConfigs(segmentDirectoryConfigs)
+ .build();
+
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig, schema);
+
+ try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
+ .load(indexDir.toURI(), segmentLoaderContext);
+ SegmentDirectory.Writer segmentWriter = segmentDirectory.createWriter()) {
+ for (IndexType indexType : postSegCreationIndexes) {
+ IndexHandler handler =
+ indexType.createIndexHandler(segmentDirectory, indexLoadingConfig.getFieldIndexConfigByColName(), schema,
+ tableConfig);
+ handler.updateIndices(segmentWriter);
+ }
+ }
+ }
+ }
+
+ /**
+ * Compute CRC and creation time, and persist to segment metadata file.
+ *
+ * @param indexDir Segment index directory
+ * @throws IOException If writing metadata fails
+ */
+ private void persistCreationMeta(File indexDir)
+ throws IOException {
+ long crc = CrcUtils.forAllFilesInFolder(indexDir).computeCrc();
+ long creationTime;
+ String creationTimeInConfig = _config.getCreationTime();
+ if (creationTimeInConfig != null) {
+ try {
+ creationTime = Long.parseLong(creationTimeInConfig);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while parsing creation time in config, use current time as creation time");
+ creationTime = System.currentTimeMillis();
+ }
+ } else {
+ creationTime = System.currentTimeMillis();
+ }
+ File segmentDir = SegmentDirectoryPaths.findSegmentDirectory(indexDir);
+ File creationMetaFile = new File(segmentDir, V1Constants.SEGMENT_CREATION_META);
+ try (DataOutputStream output = new DataOutputStream(new FileOutputStream(creationMetaFile))) {
+ output.writeLong(crc);
+ output.writeLong(creationTime);
+ }
+ }
+
@Override
public void close()
throws IOException {
- List<Closeable> creators =
- _creatorsByColAndIndex.values().stream().flatMap(map -> map.values().stream()).collect(Collectors.toList());
- creators.addAll(_nullValueVectorCreatorMap.values());
- creators.addAll(_dictionaryCreatorMap.values());
- FileUtils.close(creators);
+ closeColIndexes();
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnIndexCreators.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnIndexCreators.java
new file mode 100644
index 00000000000..1fdaeefa5ae
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnIndexCreators.java
@@ -0,0 +1,120 @@
+/**
+ * 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.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator;
+import org.apache.pinot.segment.spi.index.IndexCreator;
+import org.apache.pinot.spi.data.FieldSpec;
+
+
+/**
+ * Holds all the index creators and metadata for a single column during segment creation.
+ * This is used by ColumnarSegmentCreator to avoid hashmap lookups when processing columns.
+ */
+public class ColumnIndexCreators implements Closeable {
+ private final String _columnName;
+ private final FieldSpec _fieldSpec;
+ private final SegmentDictionaryCreator _dictionaryCreator; // null for raw columns
+ // Indexes whose build lifecycle is not DURING_SEGMENT_CREATION are not included
+ private final List<IndexCreator> _indexCreators;
+ private final NullValueVectorCreator _nullValueVectorCreator; // null if column is not nullable
+ private final boolean _isDictionaryEncoded;
+ private volatile boolean _isSealed = false;
+ private volatile boolean _isClosed = false;
+
+ public ColumnIndexCreators(String columnName, FieldSpec fieldSpec,
+ @Nullable SegmentDictionaryCreator dictionaryCreator,
+ List<IndexCreator> indexCreators,
+ @Nullable NullValueVectorCreator nullValueVectorCreator) {
+ _columnName = columnName;
+ _fieldSpec = fieldSpec;
+ _dictionaryCreator = dictionaryCreator;
+ _indexCreators = indexCreators;
+ _nullValueVectorCreator = nullValueVectorCreator;
+ _isDictionaryEncoded = dictionaryCreator != null;
+ }
+
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ public FieldSpec getFieldSpec() {
+ return _fieldSpec;
+ }
+
+ @Nullable
+ public SegmentDictionaryCreator getDictionaryCreator() {
+ return _dictionaryCreator;
+ }
+
+ public List<IndexCreator> getIndexCreators() {
+ return _indexCreators;
+ }
+
+ @Nullable
+ public NullValueVectorCreator getNullValueVectorCreator() {
+ return _nullValueVectorCreator;
+ }
+
+ public boolean isDictionaryEncoded() {
+ return _isDictionaryEncoded;
+ }
+
+ public void seal() throws IOException {
+ if (_isSealed) {
+ return;
+ }
+ // Set isSealed before sealing the creators to avoid sealing again if any exception is thrown during seal
+ _isSealed = true;
+ if (_dictionaryCreator != null) {
+ _dictionaryCreator.seal();
+ }
+ if (_nullValueVectorCreator != null) {
+ _nullValueVectorCreator.seal();
+ }
+ for (IndexCreator indexCreator : _indexCreators) {
+ indexCreator.seal();
+ }
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ if (_isClosed) {
+ return;
+ }
+ // Set isClosed before closing the creators to avoid closing again if any exception is thrown during close
+ _isClosed = true;
+ List<Closeable> creators = new ArrayList<>();
+ if (_dictionaryCreator != null) {
+ creators.add(_dictionaryCreator);
+ }
+ if (_nullValueVectorCreator != null) {
+ creators.add(_nullValueVectorCreator);
+ }
+ creators.addAll(_indexCreators);
+ FileUtils.close(creators);
+ }
+}
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 5e53464ad44..c414a32ec87 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
@@ -19,298 +19,48 @@
package org.apache.pinot.segment.local.segment.creator.impl;
import com.fasterxml.jackson.core.JsonParseException;
-import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
-import java.io.Closeable;
import java.io.File;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
-import org.apache.commons.configuration2.PropertiesConfiguration;
-import org.apache.commons.configuration2.ex.ConfigurationException;
-import org.apache.pinot.common.utils.FileUtils;
-import org.apache.pinot.segment.local.io.util.PinotDataBitSet;
import org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator;
-import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexPlugin;
-import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
import org.apache.pinot.segment.local.segment.index.forward.ForwardIndexType;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.creator.ColumnIndexCreationInfo;
-import org.apache.pinot.segment.spi.creator.IndexCreationContext;
-import org.apache.pinot.segment.spi.creator.SegmentCreator;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
-import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
-import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
-import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.segment.spi.index.IndexCreator;
-import org.apache.pinot.segment.spi.index.IndexService;
-import org.apache.pinot.segment.spi.index.IndexType;
-import org.apache.pinot.segment.spi.index.StandardIndexes;
-import org.apache.pinot.segment.spi.index.TextIndexConfig;
-import org.apache.pinot.segment.spi.index.creator.ForwardIndexCreator;
import org.apache.pinot.segment.spi.index.creator.SegmentIndexCreationInfo;
import org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
-import org.apache.pinot.segment.spi.partition.PartitionFunction;
-import org.apache.pinot.spi.config.table.IndexConfig;
-import org.apache.pinot.spi.config.table.SegmentZKPropsConfig;
-import org.apache.pinot.spi.data.ComplexFieldSpec;
-import org.apache.pinot.spi.data.DateTimeFieldSpec;
-import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.config.instance.InstanceType;
import org.apache.pinot.spi.data.FieldSpec;
-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;
-import org.joda.time.DateTimeZone;
-import org.joda.time.Interval;
-import org.joda.time.format.DateTimeFormatter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.*;
-import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Segment.*;
/**
* Segment creator which writes data in a columnar form.
*/
// TODO: check resource leaks
-public class SegmentColumnarIndexCreator implements SegmentCreator {
- // TODO Refactor class name to match interface name
- private static final Logger LOGGER = LoggerFactory.getLogger(SegmentColumnarIndexCreator.class);
- // Allow at most 512 characters for the metadata property
- static final int METADATA_PROPERTY_LENGTH_LIMIT = 512;
- private SegmentGeneratorConfig _config;
- private TreeMap<String, ColumnIndexCreationInfo> _indexCreationInfoMap;
- private final Map<String, SegmentDictionaryCreator> _dictionaryCreatorMap = new HashMap<>();
- /**
- * Contains, indexed by column name, the creator associated with each index type.
- *
- * Indexes whose build lifecycle is not DURING_SEGMENT_CREATION are not included here.
- */
- private Map<String, Map<IndexType<?, ?, ?>, IndexCreator>> _creatorsByColAndIndex = new HashMap<>();
- private final Map<String, NullValueVectorCreator> _nullValueVectorCreatorMap = new HashMap<>();
- private String _segmentName;
- private Schema _schema;
- private File _indexDir;
- private int _totalDocs;
+public class SegmentColumnarIndexCreator extends BaseSegmentCreator {
private int _docIdCounter;
@Override
public void init(SegmentGeneratorConfig segmentCreationSpec, SegmentIndexCreationInfo segmentIndexCreationInfo,
TreeMap<String, ColumnIndexCreationInfo> indexCreationInfoMap, Schema schema, File outDir,
- @Nullable int[] immutableToMutableIdMap)
+ @Nullable int[] immutableToMutableIdMap, @Nullable InstanceType instanceType)
throws Exception {
_docIdCounter = 0;
- _config = segmentCreationSpec;
- _indexCreationInfoMap = indexCreationInfoMap;
-
- // Check that the output directory does not exist
- Preconditions.checkState(!outDir.exists(), "Segment output directory: %s already exists", outDir);
-
- Preconditions.checkState(outDir.mkdirs(), "Failed to create output directory: %s", outDir);
- _indexDir = outDir;
-
- _schema = schema;
- _totalDocs = segmentIndexCreationInfo.getTotalDocs();
- if (_totalDocs == 0) {
- return;
- }
-
- Map<String, FieldIndexConfigs> indexConfigs = segmentCreationSpec.getIndexConfigsByColName();
-
- _creatorsByColAndIndex = Maps.newHashMapWithExpectedSize(indexConfigs.keySet().size());
-
- for (String columnName : indexConfigs.keySet()) {
- FieldSpec fieldSpec = schema.getFieldSpecFor(columnName);
- Preconditions.checkState(fieldSpec != null, "Failed to find column: %s in the schema", columnName);
- if (fieldSpec.isVirtualColumn()) {
- LOGGER.warn("Ignoring index creation for virtual column {}", columnName);
- continue;
- }
-
- FieldIndexConfigs originalConfig = indexConfigs.get(columnName);
- ColumnIndexCreationInfo columnIndexCreationInfo = indexCreationInfoMap.get(columnName);
- Preconditions.checkNotNull(columnIndexCreationInfo, "Missing index creation info for column: %s", columnName);
- boolean dictEnabledColumn = createDictionaryForColumn(columnIndexCreationInfo, segmentCreationSpec, fieldSpec);
- if (originalConfig.getConfig(StandardIndexes.inverted()).isEnabled()) {
- Preconditions.checkState(dictEnabledColumn,
- "Cannot create inverted index for raw index column: %s", columnName);
- }
-
- IndexType<ForwardIndexConfig, ?, ForwardIndexCreator> forwardIdx = StandardIndexes.forward();
- boolean forwardIndexDisabled = !originalConfig.getConfig(forwardIdx).isEnabled();
-
- //@formatter:off
- IndexCreationContext.Common context = IndexCreationContext.builder()
- .withIndexDir(_indexDir)
- .withDictionary(dictEnabledColumn)
- .withFieldSpec(fieldSpec)
- .withTotalDocs(segmentIndexCreationInfo.getTotalDocs())
- .withColumnIndexCreationInfo(columnIndexCreationInfo)
- .withOptimizedDictionary(_config.isOptimizeDictionary()
- || _config.isOptimizeDictionaryForMetrics() && fieldSpec.getFieldType() == FieldSpec.FieldType.METRIC)
- .onHeap(segmentCreationSpec.isOnHeap())
- .withForwardIndexDisabled(forwardIndexDisabled)
- .withTextCommitOnClose(true)
- .withImmutableToMutableIdMap(immutableToMutableIdMap)
- .withRealtimeConversion(segmentCreationSpec.isRealtimeConversion())
- .withConsumerDir(segmentCreationSpec.getConsumerDir())
- .withTableNameWithType(_config.getTableConfig().getTableName())
- .withContinueOnError(_config.isContinueOnError())
- .build();
- //@formatter:on
-
- FieldIndexConfigs config = adaptConfig(columnName, originalConfig, columnIndexCreationInfo, segmentCreationSpec);
-
- if (dictEnabledColumn) {
- // Create dictionary-encoded index
- // Initialize dictionary creator
- // TODO: Dictionary creator holds all unique values on heap. Consider keeping dictionary instead of creator
- // which uses off-heap memory.
-
- DictionaryIndexConfig dictConfig = config.getConfig(StandardIndexes.dictionary());
- if (!dictConfig.isEnabled()) {
- LOGGER.info("Creating dictionary index in column {}.{} even when it is disabled in config",
- segmentCreationSpec.getTableName(), columnName);
- }
-
- // override dictionary type if configured to do so
- if (_config.isOptimizeDictionaryType()) {
- LOGGER.info("Overriding dictionary type for column: {} using var-length dictionary: {}", columnName,
- columnIndexCreationInfo.isUseVarLengthDictionary());
- dictConfig = new DictionaryIndexConfig(dictConfig, columnIndexCreationInfo.isUseVarLengthDictionary());
- }
-
- SegmentDictionaryCreator creator =
- new DictionaryIndexPlugin().getIndexType().createIndexCreator(context, dictConfig);
-
- try {
- creator.build(context.getSortedUniqueElementsArray());
- } catch (Exception e) {
- LOGGER.error("Error building dictionary for field: {}, cardinality: {}, number of bytes per entry: {}",
- context.getFieldSpec().getName(), context.getCardinality(), creator.getNumBytesPerEntry());
- throw e;
- }
-
- _dictionaryCreatorMap.put(columnName, creator);
- }
-
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex =
- Maps.newHashMapWithExpectedSize(IndexService.getInstance().getAllIndexes().size());
- for (IndexType<?, ?, ?> index : IndexService.getInstance().getAllIndexes()) {
- if (index.getIndexBuildLifecycle() != IndexType.BuildLifecycle.DURING_SEGMENT_CREATION) {
- continue;
- }
- tryCreateIndexCreator(creatorsByIndex, index, context, config);
- }
- // TODO: Remove this when values stored as ForwardIndex stop depending on TextIndex config
- IndexCreator oldFwdCreator = creatorsByIndex.get(forwardIdx);
- if (oldFwdCreator != null) {
- Object fakeForwardValue = calculateRawValueForTextIndex(dictEnabledColumn, config, fieldSpec);
- if (fakeForwardValue != null) {
- ForwardIndexCreator castedOldFwdCreator = (ForwardIndexCreator) oldFwdCreator;
- SameValueForwardIndexCreator fakeValueFwdCreator =
- new SameValueForwardIndexCreator(fakeForwardValue, castedOldFwdCreator);
- creatorsByIndex.put(forwardIdx, fakeValueFwdCreator);
- }
- }
- _creatorsByColAndIndex.put(columnName, creatorsByIndex);
- }
-
- // Although NullValueVector is implemented as an index, it needs to be treated in a different way than other indexes
- for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
- String columnName = fieldSpec.getName();
- if (isNullable(fieldSpec)) {
- // Initialize Null value vector map
- LOGGER.info("Column: {} is nullable", columnName);
- _nullValueVectorCreatorMap.put(columnName, new NullValueVectorCreator(_indexDir, columnName));
- } else {
- LOGGER.info("Column: {} is not nullable", columnName);
- }
- }
- }
-
- private boolean isNullable(FieldSpec fieldSpec) {
- return _schema.isEnableColumnBasedNullHandling() ? fieldSpec.isNullable() : _config.isDefaultNullHandlingEnabled();
- }
-
- private FieldIndexConfigs adaptConfig(String columnName, FieldIndexConfigs config,
- ColumnIndexCreationInfo columnIndexCreationInfo, SegmentGeneratorConfig segmentCreationSpec) {
- FieldIndexConfigs.Builder builder = new FieldIndexConfigs.Builder(config);
- // Sorted columns treat the 'forwardIndexDisabled' flag as a no-op
- ForwardIndexConfig fwdConfig = config.getConfig(StandardIndexes.forward());
- if (!fwdConfig.isEnabled() && columnIndexCreationInfo.isSorted()) {
- builder.add(StandardIndexes.forward(),
- new ForwardIndexConfig.Builder(fwdConfig).withLegacyProperties(segmentCreationSpec.getColumnProperties(),
- columnName).build());
- }
- // Initialize inverted index creator; skip creating inverted index if sorted
- if (columnIndexCreationInfo.isSorted()) {
- builder.undeclare(StandardIndexes.inverted());
- }
- return builder.build();
- }
-
- /**
- * Creates the {@link IndexCreator} in a type safe way.
- *
- * This code needs to be in a specific method instead of inlined in the main loop in order to be able to use the
- * limited generic capabilities of Java.
- */
- private <C extends IndexConfig> void tryCreateIndexCreator(Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex,
- IndexType<C, ?, ?> index, IndexCreationContext.Common context, FieldIndexConfigs fieldIndexConfigs)
- throws Exception {
- C config = fieldIndexConfigs.getConfig(index);
- if (config.isEnabled()) {
- creatorsByIndex.put(index, index.createIndexCreator(context, config));
- }
- }
-
- /**
- * Returns true if dictionary should be created for a column, false otherwise.
- * Currently there are two sources for this config:
- * <ul>
- * <li> ColumnIndexCreationInfo (this is currently hard-coded to always return dictionary). </li>
- * <li> SegmentGeneratorConfig</li>
- * </ul>
- *
- * This method gives preference to the SegmentGeneratorConfig first.
- *
- * @param info Column index creation info
- * @param config Segment generation config
- * @param spec Field spec for the column
- * @return True if dictionary should be created for the column, false otherwise
- */
- private boolean createDictionaryForColumn(ColumnIndexCreationInfo info, SegmentGeneratorConfig config,
- FieldSpec spec) {
- if (spec instanceof ComplexFieldSpec) {
- return false;
- }
-
- String column = spec.getName();
- FieldIndexConfigs fieldIndexConfigs = config.getIndexConfigsByColName().get(column);
- if (fieldIndexConfigs.getConfig(StandardIndexes.dictionary()).isDisabled()) {
- return false;
- }
-
- return DictionaryIndexType.ignoreDictionaryOverride(config.isOptimizeDictionary(),
- config.isOptimizeDictionaryForMetrics(), config.getNoDictionarySizeRatioThreshold(),
- config.getNoDictionaryCardinalityRatioThreshold(), spec, fieldIndexConfigs, info.getDistinctValueCount(),
- info.getTotalNumberOfEntries());
+ Map<String, ColumnIndexCreators> colIndexes = Maps.newHashMapWithExpectedSize(
+ segmentCreationSpec.getIndexConfigsByColName().size());
+ initializeCommon(segmentCreationSpec, segmentIndexCreationInfo, indexCreationInfoMap,
+ schema, outDir, colIndexes, immutableToMutableIdMap, instanceType);
}
/**
@@ -324,18 +74,19 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
@Override
public void indexRow(GenericRow row)
throws IOException {
- for (Map.Entry<String, Map<IndexType<?, ?, ?>, IndexCreator>> byColEntry : _creatorsByColAndIndex.entrySet()) {
+ for (Map.Entry<String, ColumnIndexCreators> byColEntry : _colIndexes.entrySet()) {
String columnName = byColEntry.getKey();
+ ColumnIndexCreators colIndexes = byColEntry.getValue();
Object columnValueToIndex = row.getValue(columnName);
if (columnValueToIndex == null) {
throw new RuntimeException("Null value for column:" + columnName);
}
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex = byColEntry.getValue();
+ List<IndexCreator> creatorsByIndex = colIndexes.getIndexCreators();
FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
+ SegmentDictionaryCreator dictionaryCreator = colIndexes.getDictionaryCreator();
try {
if (fieldSpec.isSingleValueField()) {
indexSingleValueRow(dictionaryCreator, columnValueToIndex, creatorsByIndex);
@@ -345,12 +96,12 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
} catch (JsonParseException jpe) {
throw new ColumnJsonParserException(columnName, jpe);
}
- }
- for (Map.Entry<String, NullValueVectorCreator> entry : _nullValueVectorCreatorMap.entrySet()) {
// If row has null value for given column name, add to null value vector
- if (row.isNullValue(entry.getKey())) {
- entry.getValue().setNull(_docIdCounter);
+ if (colIndexes.getNullValueVectorCreator() != null) {
+ if (row.isNullValue(columnName)) {
+ colIndexes.getNullValueVectorCreator().setNull(_docIdCounter);
+ }
}
}
@@ -390,17 +141,14 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
return;
}
try (PinotSegmentColumnReader colReader = new PinotSegmentColumnReader(segment, columnName)) {
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex = _creatorsByColAndIndex.get(columnName);
- NullValueVectorCreator nullVec = _nullValueVectorCreatorMap.get(columnName);
FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
+ ColumnIndexCreators colIndexes = _colIndexes.get(columnName);
if (sortedDocIds != null) {
int onDiskDocId = 0;
for (int docId : sortedDocIds) {
// If validDodIds are provided, only index column if it's a valid doc
if (validDocIds == null || validDocIds.contains(docId)) {
- indexColumnValue(colReader, creatorsByIndex, columnName, fieldSpec, dictionaryCreator, docId, onDiskDocId,
- nullVec);
+ indexColumnValue(colReader, columnName, fieldSpec, colIndexes, docId, onDiskDocId);
onDiskDocId++;
}
}
@@ -409,8 +157,7 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
for (int docId = 0; docId < numDocs; docId++) {
// If validDodIds are provided, only index column if it's a valid doc
if (validDocIds == null || validDocIds.contains(docId)) {
- indexColumnValue(colReader, creatorsByIndex, columnName, fieldSpec, dictionaryCreator, docId, onDiskDocId,
- nullVec);
+ indexColumnValue(colReader, columnName, fieldSpec, colIndexes, docId, onDiskDocId);
onDiskDocId++;
}
}
@@ -429,10 +176,11 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
@Override
public void indexColumn(String columnName, ColumnReader columnReader)
throws IOException {
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex = _creatorsByColAndIndex.get(columnName);
- NullValueVectorCreator nullVec = _nullValueVectorCreatorMap.get(columnName);
+ List<IndexCreator> creatorsByIndex = _colIndexes.get(columnName).getIndexCreators();
+ NullValueVectorCreator nullVec = _colIndexes.get(columnName).getNullValueVectorCreator();
FieldSpec fieldSpec = _schema.getFieldSpecFor(columnName);
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(columnName);
+ SegmentDictionaryCreator dictionaryCreator = _colIndexes.get(columnName).getDictionaryCreator();
+
Object defaultNullValue = fieldSpec.getDefaultNullValue();
Object reuseColumnValueToIndex;
@@ -466,10 +214,12 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
}
private void indexColumnValue(PinotSegmentColumnReader colReader,
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex, String columnName, FieldSpec fieldSpec,
- SegmentDictionaryCreator dictionaryCreator, int sourceDocId, int onDiskDocPos,
- @Nullable NullValueVectorCreator nullVec)
+ String columnName, FieldSpec fieldSpec,
+ ColumnIndexCreators colIndexes, int sourceDocId, int onDiskDocPos)
throws IOException {
+ List<IndexCreator> creatorsByIndex = colIndexes.getIndexCreators();
+ NullValueVectorCreator nullVec = colIndexes.getNullValueVectorCreator();
+ SegmentDictionaryCreator dictionaryCreator = colIndexes.getDictionaryCreator();
Object columnValueToIndex = colReader.getValue(sourceDocId);
if (columnValueToIndex == null) {
throw new RuntimeException("Null value for column:" + columnName);
@@ -489,395 +239,20 @@ public class SegmentColumnarIndexCreator implements SegmentCreator {
}
private void indexSingleValueRow(SegmentDictionaryCreator dictionaryCreator, Object value,
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex)
+ List<IndexCreator> creatorsByIndex)
throws IOException {
int dictId = dictionaryCreator != null ? dictionaryCreator.indexOfSV(value) : -1;
- for (IndexCreator creator : creatorsByIndex.values()) {
+ for (IndexCreator creator : creatorsByIndex) {
creator.add(value, dictId);
}
}
private void indexMultiValueRow(SegmentDictionaryCreator dictionaryCreator, Object[] values,
- Map<IndexType<?, ?, ?>, IndexCreator> creatorsByIndex)
+ List<IndexCreator> creatorsByIndex)
throws IOException {
int[] dictId = dictionaryCreator != null ? dictionaryCreator.indexOfMV(values) : null;
- for (IndexCreator creator : creatorsByIndex.values()) {
+ for (IndexCreator creator : creatorsByIndex) {
creator.add(values, dictId);
}
}
-
- @Nullable
- private Object calculateRawValueForTextIndex(boolean dictEnabledColumn, FieldIndexConfigs configs,
- FieldSpec fieldSpec) {
- if (dictEnabledColumn) {
- return null;
- }
- TextIndexConfig textConfig = configs.getConfig(StandardIndexes.text());
- if (!textConfig.isEnabled()) {
- return null;
- }
-
- Object rawValue = textConfig.getRawValueForTextIndex();
-
- if (rawValue == null) {
- return null;
- } else if (!fieldSpec.isSingleValueField()) {
- if (fieldSpec.getDataType().getStoredType() == FieldSpec.DataType.STRING) {
- if (!(rawValue instanceof String[])) {
- rawValue = new String[]{String.valueOf(rawValue)};
- }
- } else if (fieldSpec.getDataType().getStoredType() == FieldSpec.DataType.BYTES) {
- if (!(rawValue instanceof String[])) {
- rawValue = new byte[][]{String.valueOf(rawValue).getBytes(StandardCharsets.UTF_8)};
- }
- } else {
- throw new RuntimeException("Text Index is only supported for STRING and BYTES stored type");
- }
- }
- return rawValue;
- }
-
- @Override
- public void setSegmentName(String segmentName) {
- _segmentName = segmentName;
- }
-
- @Override
- public void seal()
- throws ConfigurationException, IOException {
- for (SegmentDictionaryCreator creator : _dictionaryCreatorMap.values()) {
- creator.seal();
- }
- for (NullValueVectorCreator creator : _nullValueVectorCreatorMap.values()) {
- creator.seal();
- }
- for (Map<IndexType<?, ?, ?>, IndexCreator> creatorsByType : _creatorsByColAndIndex.values()) {
- for (IndexCreator creator : creatorsByType.values()) {
- creator.seal();
- }
- }
- writeMetadata();
- }
-
- private void writeMetadata()
- throws ConfigurationException {
- File metadataFile = new File(_indexDir, V1Constants.MetadataKeys.METADATA_FILE_NAME);
- PropertiesConfiguration properties = CommonsConfigurationUtils.fromFile(metadataFile);
-
- properties.setProperty(SEGMENT_CREATOR_VERSION, _config.getCreatorVersion());
- properties.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR));
- properties.setProperty(SEGMENT_NAME, _segmentName);
- properties.setProperty(TABLE_NAME, _config.getTableName());
- properties.setProperty(DIMENSIONS, _config.getDimensions());
- properties.setProperty(METRICS, _config.getMetrics());
- properties.setProperty(DATETIME_COLUMNS, _config.getDateTimeColumnNames());
- properties.setProperty(COMPLEX_COLUMNS, _config.getComplexColumnNames());
- String timeColumnName = _config.getTimeColumnName();
- properties.setProperty(TIME_COLUMN_NAME, timeColumnName);
- properties.setProperty(SEGMENT_TOTAL_DOCS, String.valueOf(_totalDocs));
-
- // Write time related metadata (start time, end time, time unit)
- if (timeColumnName != null) {
- ColumnIndexCreationInfo timeColumnIndexCreationInfo = _indexCreationInfoMap.get(timeColumnName);
- if (timeColumnIndexCreationInfo != null) {
- try {
- long startTime;
- long endTime;
- TimeUnit timeUnit;
-
- // Use start/end time in config if defined
- if (_config.getStartTime() != null) {
- startTime = Long.parseLong(_config.getStartTime());
- endTime = Long.parseLong(_config.getEndTime());
- timeUnit = Preconditions.checkNotNull(_config.getSegmentTimeUnit());
- } else {
- if (_totalDocs > 0) {
- String startTimeStr = timeColumnIndexCreationInfo.getMin().toString();
- String endTimeStr = timeColumnIndexCreationInfo.getMax().toString();
-
- if (_config.getTimeColumnType() == SegmentGeneratorConfig.TimeColumnType.SIMPLE_DATE) {
- // For TimeColumnType.SIMPLE_DATE_FORMAT, convert time value into millis since epoch
- // Use DateTimeFormatter from DateTimeFormatSpec to handle default time zone consistently.
- DateTimeFormatSpec formatSpec = _config.getDateTimeFormatSpec();
- Preconditions.checkNotNull(formatSpec, "DateTimeFormatSpec must exist for SimpleDate");
- DateTimeFormatter dateTimeFormatter = formatSpec.getDateTimeFormatter();
- startTime = dateTimeFormatter.parseMillis(startTimeStr);
- endTime = dateTimeFormatter.parseMillis(endTimeStr);
- timeUnit = TimeUnit.MILLISECONDS;
- } else {
- // by default, time column type is TimeColumnType.EPOCH
- startTime = Long.parseLong(startTimeStr);
- endTime = Long.parseLong(endTimeStr);
- timeUnit = Preconditions.checkNotNull(_config.getSegmentTimeUnit());
- }
- } else {
- // No records in segment. Use current time as start/end
- long now = System.currentTimeMillis();
- if (_config.getTimeColumnType() == SegmentGeneratorConfig.TimeColumnType.SIMPLE_DATE) {
- startTime = now;
- endTime = now;
- timeUnit = TimeUnit.MILLISECONDS;
- } else {
- timeUnit = Preconditions.checkNotNull(_config.getSegmentTimeUnit());
- startTime = timeUnit.convert(now, TimeUnit.MILLISECONDS);
- endTime = timeUnit.convert(now, TimeUnit.MILLISECONDS);
- }
- }
- }
-
- if (!_config.isSkipTimeValueCheck()) {
- Interval timeInterval =
- new Interval(timeUnit.toMillis(startTime), timeUnit.toMillis(endTime), DateTimeZone.UTC);
- Preconditions.checkState(TimeUtils.isValidTimeInterval(timeInterval),
- "Invalid segment start/end time: %s (in millis: %s/%s) for time column: %s, must be between: %s",
- timeInterval, timeInterval.getStartMillis(), timeInterval.getEndMillis(), timeColumnName,
- TimeUtils.VALID_TIME_INTERVAL);
- }
-
- properties.setProperty(SEGMENT_START_TIME, startTime);
- properties.setProperty(SEGMENT_END_TIME, endTime);
- properties.setProperty(TIME_UNIT, timeUnit);
- } catch (Exception e) {
- if (!_config.isContinueOnError()) {
- throw e;
- }
- TimeUnit timeUnit;
- long now = System.currentTimeMillis();
- long convertedStartTime;
- long convertedEndTime;
- if (_config.getTimeColumnType() == SegmentGeneratorConfig.TimeColumnType.SIMPLE_DATE) {
- convertedEndTime = now;
- convertedStartTime = TimeUtils.getValidMinTimeMillis();
- timeUnit = TimeUnit.MILLISECONDS;
- } else {
- timeUnit = _config.getSegmentTimeUnit();
- if (timeUnit != null) {
- convertedEndTime = timeUnit.convert(now, TimeUnit.MILLISECONDS);
- convertedStartTime = timeUnit.convert(TimeUtils.getValidMinTimeMillis(), TimeUnit.MILLISECONDS);
- } else {
- // Use millis as the time unit if not able to infer from config
- timeUnit = TimeUnit.MILLISECONDS;
- convertedEndTime = now;
- convertedStartTime = TimeUtils.getValidMinTimeMillis();
- }
- }
- LOGGER.warn(
- "Caught exception while writing time metadata for segment: {}, time column: {}, total docs: {}. "
- + "Continuing using current time ({}) as the end time, and min valid time ({}) as the start time "
- + "for the segment (time unit: {}).",
- _segmentName, timeColumnName, _totalDocs, convertedEndTime, convertedStartTime, timeUnit, e);
- properties.setProperty(SEGMENT_START_TIME, convertedStartTime);
- properties.setProperty(SEGMENT_END_TIME, convertedEndTime);
- properties.setProperty(TIME_UNIT, timeUnit);
- }
- }
- }
-
- for (Map.Entry<String, String> entry : _config.getCustomProperties().entrySet()) {
- properties.setProperty(entry.getKey(), entry.getValue());
- }
-
- for (Map.Entry<String, ColumnIndexCreationInfo> entry : _indexCreationInfoMap.entrySet()) {
- String column = entry.getKey();
- ColumnIndexCreationInfo columnIndexCreationInfo = entry.getValue();
- SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(column);
- int dictionaryElementSize = (dictionaryCreator != null) ? dictionaryCreator.getNumBytesPerEntry() : 0;
- addColumnMetadataInfo(properties, column, columnIndexCreationInfo, _totalDocs, _schema.getFieldSpecFor(column),
- dictionaryCreator != null, dictionaryElementSize);
- }
-
- SegmentZKPropsConfig segmentZKPropsConfig = _config.getSegmentZKPropsConfig();
- if (segmentZKPropsConfig != null) {
- properties.setProperty(Realtime.START_OFFSET, segmentZKPropsConfig.getStartOffset());
- properties.setProperty(Realtime.END_OFFSET, segmentZKPropsConfig.getEndOffset());
- }
-
- CommonsConfigurationUtils.saveToFile(properties, metadataFile);
- }
-
- public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column,
- ColumnIndexCreationInfo columnIndexCreationInfo, int totalDocs, FieldSpec fieldSpec, boolean hasDictionary,
- int dictionaryElementSize) {
- int cardinality = columnIndexCreationInfo.getDistinctValueCount();
- properties.setProperty(getKeyFor(column, CARDINALITY), String.valueOf(cardinality));
- properties.setProperty(getKeyFor(column, TOTAL_DOCS), String.valueOf(totalDocs));
- DataType dataType = fieldSpec.getDataType();
- properties.setProperty(getKeyFor(column, DATA_TYPE), String.valueOf(dataType));
- // TODO: When the column is raw (no dictionary), we should set BITS_PER_ELEMENT to -1 (invalid). Currently we set it
- // regardless of whether dictionary is created or not for backward compatibility because ForwardIndexHandler
- // doesn't update this value when converting a raw column to dictionary encoded.
- // Consider changing it after releasing 1.5.0.
- // See https://github.com/apache/pinot/pull/16921 for details
- properties.setProperty(getKeyFor(column, BITS_PER_ELEMENT),
- String.valueOf(PinotDataBitSet.getNumBitsPerValue(cardinality - 1)));
- FieldType fieldType = fieldSpec.getFieldType();
- properties.setProperty(getKeyFor(column, DICTIONARY_ELEMENT_SIZE), String.valueOf(dictionaryElementSize));
- properties.setProperty(getKeyFor(column, COLUMN_TYPE), String.valueOf(fieldType));
- properties.setProperty(getKeyFor(column, IS_SORTED), String.valueOf(columnIndexCreationInfo.isSorted()));
- properties.setProperty(getKeyFor(column, HAS_DICTIONARY), String.valueOf(hasDictionary));
- properties.setProperty(getKeyFor(column, IS_SINGLE_VALUED), String.valueOf(fieldSpec.isSingleValueField()));
- properties.setProperty(getKeyFor(column, MAX_MULTI_VALUE_ELEMENTS),
- String.valueOf(columnIndexCreationInfo.getMaxNumberOfMultiValueElements()));
- properties.setProperty(getKeyFor(column, TOTAL_NUMBER_OF_ENTRIES),
- String.valueOf(columnIndexCreationInfo.getTotalNumberOfEntries()));
- properties.setProperty(getKeyFor(column, IS_AUTO_GENERATED),
- String.valueOf(columnIndexCreationInfo.isAutoGenerated()));
- DataType storedType = dataType.getStoredType();
- if (storedType == DataType.STRING || storedType == DataType.BYTES) {
- properties.setProperty(getKeyFor(column, SCHEMA_MAX_LENGTH), fieldSpec.getEffectiveMaxLength());
- // TODO let's revisit writing effective maxLengthStrategy into metadata, as changing it right now may affect
- // segment's CRC value
- FieldSpec.MaxLengthExceedStrategy maxLengthStrategy = fieldSpec.getMaxLengthExceedStrategy();
- if (maxLengthStrategy != null) {
- properties.setProperty(getKeyFor(column, SCHEMA_MAX_LENGTH_EXCEED_STRATEGY),
- fieldSpec.getMaxLengthExceedStrategy());
- }
- }
-
- PartitionFunction partitionFunction = columnIndexCreationInfo.getPartitionFunction();
- if (partitionFunction != null) {
- properties.setProperty(getKeyFor(column, PARTITION_FUNCTION), partitionFunction.getName());
- properties.setProperty(getKeyFor(column, NUM_PARTITIONS), columnIndexCreationInfo.getNumPartitions());
- properties.setProperty(getKeyFor(column, PARTITION_VALUES), columnIndexCreationInfo.getPartitions());
- if (columnIndexCreationInfo.getPartitionFunctionConfig() != null) {
- for (Map.Entry<String, String> entry : columnIndexCreationInfo.getPartitionFunctionConfig().entrySet()) {
- properties.setProperty(getKeyFor(column, String.format("%s.%s", PARTITION_FUNCTION_CONFIG, entry.getKey())),
- entry.getValue());
- }
- }
- }
-
- // Datetime field
- if (fieldType == FieldType.DATE_TIME) {
- DateTimeFieldSpec dateTimeFieldSpec = (DateTimeFieldSpec) fieldSpec;
- properties.setProperty(getKeyFor(column, DATETIME_FORMAT), dateTimeFieldSpec.getFormat());
- properties.setProperty(getKeyFor(column, DATETIME_GRANULARITY), dateTimeFieldSpec.getGranularity());
- }
-
- if (fieldType != FieldType.COMPLEX) {
- // Regular (non-complex) field
- if (totalDocs > 0) {
- Object min = columnIndexCreationInfo.getMin();
- Object max = columnIndexCreationInfo.getMax();
- // NOTE:
- // Min/max could be null for real-time aggregate metrics. We don't directly call addColumnMinMaxValueInfo() to
- // avoid setting MIN_MAX_VALUE_INVALID flag, which will prevent ColumnMinMaxValueGenerator from generating them
- // when loading the segment.
- if (min != null && max != null) {
- addColumnMinMaxValueInfo(properties, column, min, max, storedType);
- }
- }
- } else {
- // Complex field
- ComplexFieldSpec complexFieldSpec = (ComplexFieldSpec) fieldSpec;
- properties.setProperty(getKeyFor(column, COMPLEX_CHILD_FIELD_NAMES),
- new ArrayList<>(complexFieldSpec.getChildFieldSpecs().keySet()));
- for (Map.Entry<String, FieldSpec> entry : complexFieldSpec.getChildFieldSpecs().entrySet()) {
- addFieldSpec(properties, ComplexFieldSpec.getFullChildName(column, entry.getKey()), entry.getValue());
- }
- }
-
- // TODO: Revisit whether we should set default null value for complex field
- String defaultNullValue = columnIndexCreationInfo.getDefaultNullValue().toString();
- if (storedType == DataType.STRING) {
- // NOTE: Do not limit length of default null value because we need exact value to determine whether the default
- // null value changes
- defaultNullValue = CommonsConfigurationUtils.replaceSpecialCharacterInPropertyValue(defaultNullValue);
- }
- if (defaultNullValue != null) {
- properties.setProperty(getKeyFor(column, DEFAULT_NULL_VALUE), defaultNullValue);
- }
- }
-
- /**
- * In order to persist complex field metadata, we need to recursively add child field specs
- * So, each complex field spec will have a property for its child field names and each child field will have its
- * own properties of the detailed field spec.
- * E.g. a COMPLEX type `intMap` of Map<String, Integer> has 2 child fields:
- * - key in STRING type and value in INT type.
- * Then we will have the following properties to define a COMPLEX field:
- * column.intMap.childFieldNames = [key, value]
- * column.intMap$$key.columnType = DIMENSION
- * column.intMap$$key.dataType = STRING
- * column.intMap$$key.isSingleValued = true
- * column.intMap$$value.columnType = DIMENSION
- * column.intMap$$value.dataType = INT
- * column.intMap$$value.isSingleValued = true
- */
- public static void addFieldSpec(PropertiesConfiguration properties, String column, FieldSpec fieldSpec) {
- properties.setProperty(getKeyFor(column, COLUMN_TYPE), String.valueOf(fieldSpec.getFieldType()));
- if (!column.equals(fieldSpec.getName())) {
- properties.setProperty(getKeyFor(column, COLUMN_NAME), String.valueOf(fieldSpec.getName()));
- }
- DataType dataType = fieldSpec.getDataType();
- properties.setProperty(getKeyFor(column, DATA_TYPE), String.valueOf(dataType));
- properties.setProperty(getKeyFor(column, IS_SINGLE_VALUED), String.valueOf(fieldSpec.isSingleValueField()));
- if (dataType.equals(DataType.STRING) || dataType.equals(DataType.BYTES) || dataType.equals(DataType.JSON)) {
- properties.setProperty(getKeyFor(column, SCHEMA_MAX_LENGTH), fieldSpec.getEffectiveMaxLength());
- FieldSpec.MaxLengthExceedStrategy maxLengthExceedStrategy = fieldSpec.getEffectiveMaxLengthExceedStrategy();
- if (maxLengthExceedStrategy != null) {
- properties.setProperty(getKeyFor(column, SCHEMA_MAX_LENGTH_EXCEED_STRATEGY), maxLengthExceedStrategy);
- }
- }
-
- // datetime field
- if (fieldSpec.getFieldType() == FieldType.DATE_TIME) {
- DateTimeFieldSpec dateTimeFieldSpec = (DateTimeFieldSpec) fieldSpec;
- properties.setProperty(getKeyFor(column, DATETIME_FORMAT), dateTimeFieldSpec.getFormat());
- properties.setProperty(getKeyFor(column, DATETIME_GRANULARITY), dateTimeFieldSpec.getGranularity());
- }
-
- // complex field
- if (fieldSpec.getFieldType() == FieldType.COMPLEX) {
- ComplexFieldSpec complexFieldSpec = (ComplexFieldSpec) fieldSpec;
- properties.setProperty(getKeyFor(column, COMPLEX_CHILD_FIELD_NAMES),
- new ArrayList<>(complexFieldSpec.getChildFieldSpecs().keySet()));
- for (Map.Entry<String, FieldSpec> entry : complexFieldSpec.getChildFieldSpecs().entrySet()) {
- addFieldSpec(properties, ComplexFieldSpec.getFullChildName(column, entry.getKey()), entry.getValue());
- }
- }
- }
-
- public static void addColumnMinMaxValueInfo(PropertiesConfiguration properties, String column,
- @Nullable Object minValue, @Nullable Object maxValue, DataType storedType) {
- String validMinValue = minValue != null ? getValidPropertyValue(minValue.toString(), storedType) : null;
- if (validMinValue != null) {
- properties.setProperty(getKeyFor(column, MIN_VALUE), validMinValue);
- }
- String validMaxValue = maxValue != null ? getValidPropertyValue(maxValue.toString(), storedType) : null;
- if (validMaxValue != null) {
- properties.setProperty(getKeyFor(column, MAX_VALUE), validMaxValue);
- }
- if (validMinValue == null && validMaxValue == null) {
- properties.setProperty(getKeyFor(column, MIN_MAX_VALUE_INVALID), true);
- }
- }
-
- /**
- * Helper method to get the valid value for setting min/max. Returns {@code null} if the value is too long (longer
- * than 512 characters), or is not supported in {@link PropertiesConfiguration}, e.g. contains character with
- * surrogate.
- */
- @Nullable
- private static String getValidPropertyValue(String value, DataType storedType) {
- if (value.length() > METADATA_PROPERTY_LENGTH_LIMIT) {
- return null;
- }
- return storedType == DataType.STRING ? CommonsConfigurationUtils.replaceSpecialCharacterInPropertyValue(value)
- : value;
- }
-
- public static void removeColumnMetadataInfo(PropertiesConfiguration properties, String column) {
- properties.subset(COLUMN_PROPS_KEY_PREFIX + column).clear();
- }
-
- @Override
- public void close()
- throws IOException {
- List<Closeable> creators =
- _creatorsByColAndIndex.values().stream().flatMap(map -> map.values().stream()).collect(Collectors.toList());
- creators.addAll(_nullValueVectorCreatorMap.values());
- creators.addAll(_dictionaryCreatorMap.values());
- FileUtils.close(creators);
- }
}
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 bf14daa512b..c5226934745 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
@@ -19,21 +19,14 @@
package org.apache.pinot.segment.local.segment.creator.impl;
import com.google.common.base.Preconditions;
-import java.io.DataOutputStream;
import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
-import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.metrics.MinionMeter;
import org.apache.pinot.common.metrics.MinionMetrics;
@@ -43,18 +36,11 @@ import org.apache.pinot.segment.local.realtime.converter.stats.RealtimeSegmentSe
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;
import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.invertedindex.MultiColumnTextIndexHandler;
import org.apache.pinot.segment.local.segment.readers.CompactedPinotSegmentRecordReader;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
-import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder;
-import org.apache.pinot.segment.local.utils.CrcUtils;
import org.apache.pinot.segment.local.utils.IngestionUtils;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.V1Constants;
-import org.apache.pinot.segment.spi.converter.SegmentFormatConverter;
import org.apache.pinot.segment.spi.creator.ColumnIndexCreationInfo;
import org.apache.pinot.segment.spi.creator.ColumnStatistics;
import org.apache.pinot.segment.spi.creator.SegmentCreationDataSource;
@@ -62,22 +48,13 @@ import org.apache.pinot.segment.spi.creator.SegmentCreator;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
import org.apache.pinot.segment.spi.creator.SegmentPreIndexStatsContainer;
-import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
-import org.apache.pinot.segment.spi.index.IndexHandler;
-import org.apache.pinot.segment.spi.index.IndexService;
-import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.creator.SegmentIndexCreationInfo;
import org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
-import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderContext;
-import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderRegistry;
-import org.apache.pinot.segment.spi.store.SegmentDirectory;
-import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
import org.apache.pinot.spi.config.instance.InstanceType;
-import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -90,9 +67,7 @@ import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.spi.data.readers.RecordReaderFactory;
-import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.ByteArray;
-import org.apache.pinot.spi.utils.ReadMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -343,7 +318,7 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
// TODO: _indexCreationInfoMap holds the reference to all unique values on heap (ColumnIndexCreationInfo ->
// ColumnStatistics) throughout the segment creation. Find a way to release the memory early.
_indexCreator.init(_config, _segmentIndexCreationInfo, _indexCreationInfoMap, _dataSchema, _tempIndexDir,
- immutableToMutableIdMap);
+ immutableToMutableIdMap, _instanceType);
// Build the index
_recordReader.rewind();
@@ -452,7 +427,7 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
// TODO: _indexCreationInfoMap holds the reference to all unique values on heap (ColumnIndexCreationInfo ->
// ColumnStatistics) throughout the segment creation. Find a way to release the memory early.
_indexCreator.init(_config, _segmentIndexCreationInfo, _indexCreationInfoMap, _dataSchema, _tempIndexDir,
- immutableToMutableIdMap);
+ immutableToMutableIdMap, _instanceType);
// Build the indexes
LOGGER.info("Start building Index by column");
@@ -480,217 +455,34 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
private void handlePostCreation()
throws Exception {
- ColumnStatistics timeColumnStatistics = _segmentStats.getColumnProfileFor(_config.getTimeColumnName());
- int sequenceId = _config.getSequenceId();
- if (timeColumnStatistics != null) {
- if (_totalDocs > 0) {
- _segmentName = _config.getSegmentNameGenerator()
- .generateSegmentName(sequenceId, timeColumnStatistics.getMinValue(), timeColumnStatistics.getMaxValue());
- } else {
- // When totalDoc is 0, check whether 'failOnEmptySegment' option is true. If so, directly fail the segment
- // creation.
- Preconditions.checkArgument(!_config.isFailOnEmptySegment(),
- "Failing the empty segment creation as the option 'failOnEmptySegment' is set to: "
- + _config.isFailOnEmptySegment());
- // Generate a unique name for a segment with no rows
- long now = System.currentTimeMillis();
- _segmentName = _config.getSegmentNameGenerator().generateSegmentName(sequenceId, now, now);
- }
- } else {
- _segmentName = _config.getSegmentNameGenerator().generateSegmentName(sequenceId, null, null);
- }
-
+ // Execute all post-creation operations directly on the index creator
try {
- // Write the index files to disk
- _indexCreator.setSegmentName(_segmentName);
_indexCreator.seal();
} finally {
_indexCreator.close();
}
- LOGGER.info("Finished segment seal!");
+ _segmentName = _indexCreator.getSegmentName();
- // Delete the directory named after the segment name, if it exists
- final File outputDir = new File(_config.getOutDir());
- final File segmentOutputDir = new File(outputDir, _segmentName);
+ // Move the segment from the temporary directory to the final output directory
+ File outputDir = new File(_config.getOutDir());
+ File segmentOutputDir = new File(outputDir, _segmentName);
if (segmentOutputDir.exists()) {
FileUtils.deleteDirectory(segmentOutputDir);
}
-
- // Move the temporary directory into its final location
FileUtils.moveDirectory(_tempIndexDir, segmentOutputDir);
-
- // Delete the temporary directory
FileUtils.deleteQuietly(_tempIndexDir);
- convertFormatIfNecessary(segmentOutputDir);
-
- if (_totalDocs > 0) {
- buildStarTreeV2IfNecessary(segmentOutputDir);
- buildMultiColumnTextIndex(segmentOutputDir);
- }
- updatePostSegmentCreationIndexes(segmentOutputDir);
-
- // Compute CRC and creation time
- long crc = CrcUtils.forAllFilesInFolder(segmentOutputDir).computeCrc();
- long creationTime;
- String creationTimeInConfig = _config.getCreationTime();
- if (creationTimeInConfig != null) {
- try {
- creationTime = Long.parseLong(creationTimeInConfig);
- } catch (Exception e) {
- LOGGER.error("Caught exception while parsing creation time in config, use current time as creation time");
- creationTime = System.currentTimeMillis();
- }
- } else {
- creationTime = System.currentTimeMillis();
- }
-
- // Persist creation metadata to disk
- persistCreationMeta(segmentOutputDir, crc, creationTime);
-
LOGGER.info("Driver, record read time (in ms) : {}", TimeUnit.NANOSECONDS.toMillis(_totalRecordReadTimeNs));
LOGGER.info("Driver, stats collector time (in ms) : {}", TimeUnit.NANOSECONDS.toMillis(_totalStatsCollectorTimeNs));
LOGGER.info("Driver, indexing time (in ms) : {}", TimeUnit.NANOSECONDS.toMillis(_totalIndexTimeNs));
}
- private void buildMultiColumnTextIndex(File segmentOutputDir)
- throws Exception {
- if (_config.getMultiColumnTextIndexConfig() != null) {
- PinotConfiguration segmentDirectoryConfigs =
- new PinotConfiguration(Map.of(IndexLoadingConfig.READ_MODE_KEY, ReadMode.mmap));
-
- TableConfig tableConfig = _config.getTableConfig();
- Schema schema = _config.getSchema();
- SegmentDirectoryLoaderContext segmentLoaderContext =
- new SegmentDirectoryLoaderContext.Builder()
- .setTableConfig(tableConfig)
- .setSchema(schema)
- .setSegmentName(_segmentName)
- .setSegmentDirectoryConfigs(segmentDirectoryConfigs)
- .build();
-
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig, schema);
-
- try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
- .load(segmentOutputDir.toURI(), segmentLoaderContext);
- SegmentDirectory.Writer segmentWriter = segmentDirectory.createWriter()) {
- MultiColumnTextIndexHandler handler = new MultiColumnTextIndexHandler(segmentDirectory, indexLoadingConfig,
- _config.getMultiColumnTextIndexConfig());
- handler.updateIndices(segmentWriter);
- handler.postUpdateIndicesCleanup(segmentWriter);
- }
- }
- }
-
- private void updatePostSegmentCreationIndexes(File indexDir)
- throws Exception {
- Set<IndexType> postSegCreationIndexes = IndexService.getInstance().getAllIndexes().stream()
- .filter(indexType -> indexType.getIndexBuildLifecycle() == IndexType.BuildLifecycle.POST_SEGMENT_CREATION)
- .collect(Collectors.toSet());
-
- if (!postSegCreationIndexes.isEmpty()) {
- // Build other indexes
- Map<String, Object> props = new HashMap<>();
- props.put(IndexLoadingConfig.READ_MODE_KEY, ReadMode.mmap);
- PinotConfiguration segmentDirectoryConfigs = new PinotConfiguration(props);
-
- TableConfig tableConfig = _config.getTableConfig();
- Schema schema = _config.getSchema();
- SegmentDirectoryLoaderContext segmentLoaderContext =
- new SegmentDirectoryLoaderContext.Builder()
- .setTableConfig(tableConfig)
- .setSchema(schema)
- .setSegmentName(_segmentName)
- .setSegmentDirectoryConfigs(segmentDirectoryConfigs)
- .build();
-
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig, schema);
-
- try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
- .load(indexDir.toURI(), segmentLoaderContext);
- SegmentDirectory.Writer segmentWriter = segmentDirectory.createWriter()) {
- for (IndexType indexType : postSegCreationIndexes) {
- IndexHandler handler =
- indexType.createIndexHandler(segmentDirectory, indexLoadingConfig.getFieldIndexConfigByColName(), schema,
- tableConfig);
- handler.updateIndices(segmentWriter);
- }
- }
- }
- }
-
- private void buildStarTreeV2IfNecessary(File indexDir)
- throws Exception {
- List<StarTreeIndexConfig> starTreeIndexConfigs = _config.getStarTreeIndexConfigs();
- boolean enableDefaultStarTree = _config.isEnableDefaultStarTree();
- if (CollectionUtils.isNotEmpty(starTreeIndexConfigs) || enableDefaultStarTree) {
- MultipleTreesBuilder.BuildMode buildMode =
- _config.isOnHeap() ? MultipleTreesBuilder.BuildMode.ON_HEAP : MultipleTreesBuilder.BuildMode.OFF_HEAP;
- MultipleTreesBuilder builder = new MultipleTreesBuilder(starTreeIndexConfigs, enableDefaultStarTree, indexDir,
- buildMode);
- // We don't create the builder using the try-with-resources pattern because builder.close() performs
- // some clean-up steps to roll back the star-tree index to the previous state if it exists. If this goes wrong
- // the star-tree index can be in an inconsistent state. To prevent that, when builder.close() throws an
- // exception we want to propagate that up instead of ignoring it. This can get clunky when using
- // try-with-resources as in this scenario the close() exception will be added to the suppressed exception list
- // rather than thrown as the main exception, even though the original exception thrown on build() is ignored.
- try {
- builder.build();
- } catch (Exception e) {
- String tableNameWithType = _config.getTableConfig().getTableName();
- LOGGER.error("Failed to build star-tree index for table: {}, skipping", tableNameWithType, e);
- if (_instanceType == InstanceType.MINION) {
- MinionMetrics.get().addMeteredTableValue(tableNameWithType, MinionMeter.STAR_TREE_INDEX_BUILD_FAILURES, 1);
- } else {
- ServerMetrics.get().addMeteredTableValue(tableNameWithType, ServerMeter.STAR_TREE_INDEX_BUILD_FAILURES, 1);
- }
- } finally {
- builder.close();
- }
- }
- }
-
- // Explanation of why we are using format converter:
- // There are 3 options to correctly generate segments to v3 format
- // 1. Generate v3 directly: This is efficient but v3 index writer needs to know buffer size upfront.
- // Inverted, star and raw indexes don't have the index size upfront. This is also least flexible approach
- // if we add more indexes in future.
- // 2. Hold data in-memory: One way to work around predeclaring sizes in (1) is to allocate "large" buffer (2GB?)
- // and hold the data in memory and write the buffer at the end. The memory requirement in this case increases linearly
- // with the number of columns. Variation of that is to mmap data to separate files...which is what we are doing here
- // 3. Another option is to generate dictionary and fwd indexes in v3 and generate inverted, star and raw indexes in
- // separate files. Then add those files to v3 index file. This leads to lot of hodgepodge code to
- // handle multiple segment formats.
- // Using converter is similar to option (2), plus it's battle-tested code. We will roll out with
- // this change to keep changes limited. Once we've migrated we can implement approach (1) with option to
- // copy for indexes for which we don't know sizes upfront.
- private void convertFormatIfNecessary(File segmentDirectory)
- throws Exception {
- SegmentVersion versionToGenerate = _config.getSegmentVersion();
- if (versionToGenerate.equals(SegmentVersion.v1)) {
- // v1 by default
- return;
- }
- SegmentFormatConverter converter = SegmentFormatConverterFactory.getConverter(SegmentVersion.v1, SegmentVersion.v3);
- converter.convert(segmentDirectory);
- }
-
@Override
public ColumnStatistics getColumnStatisticsCollector(final String columnName)
throws Exception {
return _segmentStats.getColumnProfileFor(columnName);
}
- public static void persistCreationMeta(File indexDir, long crc, long creationTime)
- throws IOException {
- File segmentDir = SegmentDirectoryPaths.findSegmentDirectory(indexDir);
- File creationMetaFile = new File(segmentDir, V1Constants.SEGMENT_CREATION_META);
- try (DataOutputStream output = new DataOutputStream(new FileOutputStream(creationMetaFile))) {
- output.writeLong(crc);
- output.writeLong(creationTime);
- }
- }
-
/**
* Complete the stats gathering process and store the stats information in indexCreationInfoMap.
*/
@@ -819,7 +611,8 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
}
// Initialize the index creation using the per-column statistics information
- _indexCreator.init(_config, _segmentIndexCreationInfo, _indexCreationInfoMap, _dataSchema, _tempIndexDir, null);
+ _indexCreator.init(_config, _segmentIndexCreationInfo, _indexCreationInfoMap, _dataSchema, _tempIndexDir,
+ null, _instanceType);
// Build the indexes column-wise (true column-major approach)
LOGGER.info("Start building Index using columnar approach");
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
index 88d829ae06d..f944254f9c2 100644
--- 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
@@ -24,6 +24,8 @@ 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.segment.spi.index.FieldIndexConfigs;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigsUtil;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.ColumnReader;
@@ -75,14 +77,16 @@ public class ColumnarSegmentPreIndexStatsContainer implements SegmentPreIndexSta
* Initialize stats collectors for all columns in the target schema.
*/
private void initializeStatsCollectors() {
+ Map<String, FieldIndexConfigs> indexConfigsByCol = FieldIndexConfigsUtil.createIndexConfigsByColName(
+ _statsCollectorConfig.getTableConfig(), _statsCollectorConfig.getSchema());
for (FieldSpec fieldSpec : _targetSchema.getAllFieldSpecs()) {
if (fieldSpec.isVirtualColumn()) {
continue;
}
String columnName = fieldSpec.getName();
- AbstractColumnStatisticsCollector collector =
- StatsCollectorUtil.createStatsCollector(columnName, fieldSpec, _statsCollectorConfig);
+ AbstractColumnStatisticsCollector collector = StatsCollectorUtil.createStatsCollector(
+ columnName, fieldSpec, indexConfigsByCol.get(columnName), _statsCollectorConfig);
_columnStatsCollectorMap.put(columnName, collector);
}
}
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 ae02b062df9..eb18fbb6ad7 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
@@ -20,13 +20,11 @@ package org.apache.pinot.segment.local.segment.creator.impl.stats;
import java.util.HashMap;
import java.util.Map;
-import org.apache.pinot.segment.local.utils.ClusterConfigForTable;
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.segment.spi.index.FieldIndexConfigs;
import org.apache.pinot.segment.spi.index.FieldIndexConfigsUtil;
-import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -54,18 +52,8 @@ public class SegmentPreIndexStatsCollectorImpl implements SegmentPreIndexStatsCo
Schema dataSchema = _statsCollectorConfig.getSchema();
for (FieldSpec fieldSpec : dataSchema.getAllFieldSpecs()) {
String column = fieldSpec.getName();
- boolean dictionaryEnabled = indexConfigsByCol.get(column).getConfig(StandardIndexes.dictionary()).isEnabled();
- if (!dictionaryEnabled) {
- // MAP collector is optimised for no-dictionary collection
- if (!fieldSpec.getDataType().getStoredType().equals(FieldSpec.DataType.MAP)) {
- if (ClusterConfigForTable.useOptimizedNoDictCollector(_statsCollectorConfig.getTableConfig())) {
- _columnStatsCollectorMap.put(column, new NoDictColumnStatisticsCollector(column, _statsCollectorConfig));
- continue;
- }
- }
- }
- AbstractColumnStatisticsCollector collector =
- StatsCollectorUtil.createStatsCollector(column, fieldSpec, _statsCollectorConfig);
+ AbstractColumnStatisticsCollector collector = StatsCollectorUtil.createStatsCollector(
+ column, fieldSpec, indexConfigsByCol.get(column), _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
index 469184c3f4b..a3e3e9a4bb3 100644
--- 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
@@ -18,7 +18,10 @@
*/
package org.apache.pinot.segment.local.segment.creator.impl.stats;
+import org.apache.pinot.segment.local.utils.ClusterConfigForTable;
import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.spi.data.FieldSpec;
@@ -40,7 +43,16 @@ public final class StatsCollectorUtil {
* @return AbstractColumnStatisticsCollector for the column
*/
public static AbstractColumnStatisticsCollector createStatsCollector(String columnName, FieldSpec fieldSpec,
- StatsCollectorConfig statsCollectorConfig) {
+ FieldIndexConfigs indexConfig, StatsCollectorConfig statsCollectorConfig) {
+ boolean dictionaryEnabled = indexConfig.getConfig(StandardIndexes.dictionary()).isEnabled();
+ if (!dictionaryEnabled) {
+ // MAP collector is optimised for no-dictionary collection
+ if (!fieldSpec.getDataType().getStoredType().equals(FieldSpec.DataType.MAP)) {
+ if (ClusterConfigForTable.useOptimizedNoDictCollector(statsCollectorConfig.getTableConfig())) {
+ return new NoDictColumnStatisticsCollector(columnName, statsCollectorConfig);
+ }
+ }
+ }
switch (fieldSpec.getDataType().getStoredType()) {
case INT:
return new IntColumnPreIndexStatsCollector(columnName, statsCollectorConfig);
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentCreator.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentCreator.java
index 41737eca187..172c696b08f 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentCreator.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentCreator.java
@@ -24,10 +24,10 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.TreeMap;
import javax.annotation.Nullable;
-import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.index.creator.SegmentIndexCreationInfo;
import org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.apache.pinot.spi.config.instance.InstanceType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.ColumnReader;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -47,11 +47,14 @@ public interface SegmentCreator extends Closeable, Serializable {
* @param indexCreationInfoMap
* @param schema
* @param outDir
+ * @param immutableToMutableIdMap
+ * @param instanceType - Instance type that's used to select the metrics for observability
+ * TODO - Move instanceType to SegmentGeneratorConfig and avoid passing it here
* @throws Exception
*/
void init(SegmentGeneratorConfig segmentCreationSpec, SegmentIndexCreationInfo segmentIndexCreationInfo,
TreeMap<String, ColumnIndexCreationInfo> indexCreationInfoMap, Schema schema, File outDir,
- @Nullable int[] immutableToMutableIdMap)
+ @Nullable int[] immutableToMutableIdMap, @Nullable InstanceType instanceType)
throws Exception;
/**
@@ -91,19 +94,19 @@ public interface SegmentCreator extends Closeable, Serializable {
indexColumn(columnName, sortedDocIds, segment);
}
- /**
- * Sets the name of the segment.
- *
- * @param segmentName The name of the segment
- */
- void setSegmentName(String segmentName);
+ String getSegmentName();
/**
- * Seals the segment, flushing it to disk.
+ * Seals and creates the final segment in outDir provided in init().
+ * This method is supposed to
+ * 1. flush all column indexes to disk
+ * 2. generate the segment name
+ * 3. convert the segment to the final format
+ * 4. build other indexes (startree index, etc.) if needed.
+ * 5. persist the segment metadata and creation info files.
*
- * @throws ConfigurationException
- * @throws IOException
+ * @throws Exception If finalization fails
*/
void seal()
- throws ConfigurationException, IOException;
+ throws Exception;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org