This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 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 bff525d98ea Keep source nulls when deriving a column at segment load
(#19955)
bff525d98ea is described below
commit bff525d98ea7b1b2120777af4ad5c48a1089677f
Author: RAGHVENDRA KUMAR YADAV <[email protected]>
AuthorDate: Sun Oct 25 13:54:33 2026 -0700
Keep source nulls when deriving a column at segment load (#19955)
* Keep source nulls when deriving a column at segment load
A column derived from ingestionConfig.transformConfigs at load read its
source through PinotSegmentColumnReader without the source's null value
vector, so a null source value reached the function as the default value
stored in its place. times(source, 2) over a null source produced a
non-null 0.0 instead of the null that ingestion produces, which changes
IS NULL filters, counts and averages on the derived column.
Open the source's null value vector when it has one and pass null to the
function for those docs, as ingestion does.
* Close derived-column readers when building one fails
---
.../defaultcolumn/BaseDefaultColumnHandler.java | 57 +++++++++++++-----
.../index/loader/SegmentPreProcessorTest.java | 67 ++++++++++++++++++++++
2 files changed, 109 insertions(+), 15 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
index 1727d8c2ba6..0486ed2a87d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
@@ -62,6 +62,7 @@ import
org.apache.pinot.segment.local.segment.index.forward.CompressionStatsMeta
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import
org.apache.pinot.segment.local.segment.index.loader.InvertedIndexAndDictionaryBasedForwardIndexCreator;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
+import
org.apache.pinot.segment.local.segment.index.readers.NullValueVectorReaderImpl;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.ClusterConfigForTable;
import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -631,15 +632,23 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
int numArguments = argumentSources.size();
List<ValueReader> valueReaders = new ArrayList<>(numArguments);
List<IntFunction<Object>> argumentValues = new ArrayList<>(numArguments);
- for (DerivedColumnArgument argumentSource : argumentSources) {
- Object[] values = argumentSource.getValues();
- if (values != null) {
- argumentValues.add(docId -> values[docId]);
- } else {
- ValueReader valueReader = new
ValueReader(argumentSource.getColumnMetadata());
- valueReaders.add(valueReader);
- argumentValues.add(valueReader::getValue);
+ try {
+ for (DerivedColumnArgument argumentSource : argumentSources) {
+ Object[] values = argumentSource.getValues();
+ if (values != null) {
+ argumentValues.add(docId -> values[docId]);
+ } else {
+ ValueReader valueReader = new
ValueReader(argumentSource.getColumnMetadata());
+ valueReaders.add(valueReader);
+ argumentValues.add(valueReader::getValue);
+ }
+ }
+ } catch (IOException | IndexReaderConstraintException | RuntimeException
e) {
+ // The finally below only runs once the readers are built, so close the
ones opened before the failure here
+ for (ValueReader valueReader : valueReaders) {
+ valueReader.close();
}
+ throw e;
}
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);
@@ -1685,17 +1694,35 @@ public abstract class BaseDefaultColumnHandler
implements DefaultColumnHandler {
.add(StandardIndexes.forward(),
ForwardIndexConfig.getDefault(columnMetadata.getForwardIndexEncoding()))
.build();
_forwardIndexReader = readerFactory.createIndexReader(_segmentWriter,
fieldIndexConfigs, columnMetadata);
- if (columnMetadata.hasDictionary()) {
- _dictionary = DictionaryIndexType.read(_segmentWriter, columnMetadata);
- } else {
- _dictionary = null;
+ Dictionary dictionary = null;
+ try {
+ if (columnMetadata.hasDictionary()) {
+ dictionary = DictionaryIndexType.read(_segmentWriter,
columnMetadata);
+ }
+ // A null source value is stored as the column's default value. Read
the null vector too, so the function sees
+ // null for those docs, as it does at ingestion, instead of deriving a
non-null value from the placeholder.
+ String column = columnMetadata.getColumnName();
+ NullValueVectorReaderImpl nullValueVectorReader = null;
+ if (_segmentWriter.hasIndexFor(column,
StandardIndexes.nullValueVector())) {
+ nullValueVectorReader =
+ new NullValueVectorReaderImpl(_segmentWriter.getIndexFor(column,
StandardIndexes.nullValueVector()));
+ }
+ _columnReader = new PinotSegmentColumnReader(column,
_forwardIndexReader, dictionary, nullValueVectorReader,
+ columnMetadata.getMaxNumberOfMultiValues());
+ } catch (IOException | RuntimeException e) {
+ // The caller never gets this reader to close, so release what was
opened before the failure
+ if (dictionary != null) {
+ dictionary.close();
+ }
+ _forwardIndexReader.close();
+ throw e;
}
- _columnReader = new
PinotSegmentColumnReader(columnMetadata.getColumnName(), _forwardIndexReader,
_dictionary,
- null, columnMetadata.getMaxNumberOfMultiValues());
+ _dictionary = dictionary;
}
+ @Nullable
Object getValue(int docId) {
- return _columnReader.getValue(docId);
+ return _columnReader.isNull(docId) ? null :
_columnReader.getValue(docId);
}
@Override
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
index ccc323e742c..81dd3e48506 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
@@ -46,6 +46,7 @@ import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationD
import
org.apache.pinot.segment.local.segment.index.converter.SegmentV1V2ToV3FormatConverter;
import
org.apache.pinot.segment.local.segment.index.loader.columnminmaxvalue.ColumnMinMaxValueGeneratorMode;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
@@ -2288,6 +2289,63 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
}
}
+ /// A column derived at load must see the source's nulls the way ingestion
does: a null source value is a null input
+ /// to the function, not the default value stored in its place, so the
derived value is null too.
+ @Test(dataProvider = "bothV1AndV3")
+ public void testDerivedColumnFromNullableSourceKeepsNulls(SegmentVersion
segmentVersion)
+ throws Exception {
+ Schema schema =
+ new
Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME).addMetric("source",
DataType.DOUBLE).build();
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNullHandlingEnabled(true).build();
+ Double[] sourceValues = {1.5, null, 3.0, null, 5.0};
+ List<GenericRow> rows = new ArrayList<>(sourceValues.length);
+ for (Double value : sourceValues) {
+ GenericRow row = new GenericRow();
+ row.putValue("source", value);
+ rows.add(row);
+ }
+ FileUtils.deleteQuietly(TEMP_DIR);
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig,
schema);
+ config.setInstanceType(InstanceType.SERVER);
+ config.setOutDir(TEMP_DIR.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME);
+ config.setSegmentVersion(segmentVersion);
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+
+ Schema derivedSchema = new
Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addMetric("source", DataType.DOUBLE)
+ .addMetric("derived", DataType.DOUBLE)
+ .build();
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setTransformConfigs(List.of(new TransformConfig("derived",
"times(source, 2)")));
+ TableConfig derivedTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNullHandlingEnabled(true)
+ .setIngestionConfig(ingestionConfig)
+ .build();
+ IndexLoadingConfig indexLoadingConfig = new
IndexLoadingConfig(derivedTableConfig, derivedSchema);
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+ SegmentPreProcessor processor = new
SegmentPreProcessor(segmentDirectory, indexLoadingConfig)) {
+ processor.process(SEGMENT_OPERATIONS_THROTTLER);
+ }
+
+ ImmutableSegment segment = ImmutableSegmentLoader.load(INDEX_DIR,
indexLoadingConfig, false);
+ try (PinotSegmentColumnReader reader = new
PinotSegmentColumnReader(segment, "derived")) {
+ for (int docId = 0; docId < sourceValues.length; docId++) {
+ if (sourceValues[docId] == null) {
+ assertTrue(reader.isNull(docId), "Derived value must be null where
the source is null, doc: " + docId);
+ } else {
+ assertFalse(reader.isNull(docId), "Derived value must not be null,
doc: " + docId);
+ assertEquals(reader.getValue(docId), sourceValues[docId] * 2);
+ }
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+
private void buildStarTreeTestSegment(TableConfig tableConfig, Schema schema)
throws Exception {
FileUtils.deleteQuietly(TEMP_DIR);
@@ -2709,6 +2767,9 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
@Test(dataProvider = "bothV1AndV3")
public void
testDerivedColumnReadingDefaultColumnAddedInSameReload(SegmentVersion
segmentVersion)
throws Exception {
+ // With null handling on, a default column added in this reload is null on
every row, and the derived column reads
+ // that null as ingestion would. Turn it off so the source holds its
default value and the ordering is visible.
+ _nullHandlingEnabled = false;
buildSegment(segmentVersion);
// newIntMetric is NOT derived -- it is a plain new column filled with its
default value (0).
@@ -2949,6 +3010,9 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
@Test(dataProvider = "bothV1AndV3")
public void testDerivedColumnOrderedAfterUpdatedSource(SegmentVersion
segmentVersion)
throws Exception {
+ // With null handling on, a default column added in this reload is null on
every row, and the derived column reads
+ // that null as ingestion would. Turn it off so the source holds its
default value and the ordering is visible.
+ _nullHandlingEnabled = false;
buildSegment(segmentVersion);
// First reload: the source is added, holding the default value its schema
declares.
@@ -2984,6 +3048,9 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
@Test
public void
testDerivedColumnFromForwardIndexDisabledColumnAddedInSameReload()
throws Exception {
+ // With null handling on, a default column added in this reload is null on
every row, and the derived column reads
+ // that null as ingestion would. Turn it off so the source holds its
default value and the ordering is visible.
+ _nullHandlingEnabled = false;
buildSegment(SegmentVersion.v3);
// A default value of its own, so a derived column that fell back to ITS
default would be visibly different.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
