Skip to content

Regenerate derived columns when transforms change #18977

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Vamsi-klu wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Regenerate derived columns when transforms change #18977

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 42 additions & 1 deletion ...src/main/java/org/apache/pinot/segment/local/segment/creator/impl/BaseSegmentCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,8 @@
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.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DateTimeFormatSpec;
Expand Down Expand Up @@ -549,7 +551,8 @@ protected void writeMetadata()
ForwardIndexConfig fwdConfig =
columnIndexCreators.getIndexConfigs().getConfig(StandardIndexes.forward());
addColumnMetadataInfo(properties, column, columnStatistics, _totalDocs, _schema.getFieldSpecFor(column),
hasDictionary, dictionaryElementSize, fwdConfig.getEncodingType(), false);
hasDictionary, dictionaryElementSize, fwdConfig.getEncodingType(), false,
getTransformFunctionForColumn(column));
}

SegmentZKPropsConfig segmentZKPropsConfig = _config.getSegmentZKPropsConfig();
Expand All @@ -564,6 +567,15 @@ protected void writeMetadata()
public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column,
ColumnStatistics columnStatistics, int totalDocs, FieldSpec fieldSpec, boolean hasDictionary,
int dictionaryElementSize, FieldConfig.EncodingType forwardIndexEncoding, boolean autoGenerated) {
addColumnMetadataInfo(properties, column, columnStatistics, totalDocs, fieldSpec, hasDictionary,
dictionaryElementSize, forwardIndexEncoding, autoGenerated, null);
}

/// Adds column metadata information to the properties configuration.
public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column,
ColumnStatistics columnStatistics, int totalDocs, FieldSpec fieldSpec, boolean hasDictionary,
int dictionaryElementSize, FieldConfig.EncodingType forwardIndexEncoding, boolean autoGenerated,
@Nullable String transformFunction) {
addFieldSpec(properties, column, fieldSpec);
properties.setProperty(getKeyFor(column, TOTAL_DOCS), String.valueOf(totalDocs));
int cardinality = columnStatistics.getCardinality();
Expand Down Expand Up @@ -606,6 +618,7 @@ public static void addColumnMetadataInfo(PropertiesConfiguration properties, Str
if (autoGenerated) {
properties.setProperty(getKeyFor(column, IS_AUTO_GENERATED), "true");
}
addTransformFunction(properties, column, transformFunction);

// Min/max value
if (fieldSpec.getFieldType() != FieldType.COMPLEX) {
Expand Down Expand Up @@ -639,6 +652,34 @@ public static void addColumnMetadataInfo(PropertiesConfiguration properties, Str
}
}

private static void addTransformFunction(PropertiesConfiguration properties, String column,
@Nullable String transformFunction) {
if (transformFunction != null) {
String validTransformFunction =
CommonsConfigurationUtils.replaceSpecialCharacterInPropertyValue(transformFunction);
if (validTransformFunction != null) {
properties.setProperty(getKeyFor(column, TRANSFORM_FUNCTION), validTransformFunction);
}
}
}

@Nullable
@SuppressWarnings("deprecation")
private String getTransformFunctionForColumn(String column) {
TableConfig tableConfig = _config.getTableConfig();
IngestionConfig ingestionConfig = tableConfig != null ? tableConfig.getIngestionConfig() : null;
if (ingestionConfig != null && ingestionConfig.getTransformConfigs() != null) {
for (TransformConfig transformConfig : ingestionConfig.getTransformConfigs()) {
if (column.equals(transformConfig.getColumnName())) {
return transformConfig.getTransformFunction();
}
}
}
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);
// Keep the schema-level transform fallback for segments built from legacy configs.
return fieldSpec != null ? fieldSpec.getTransformFunction() : null;
}

/**
* 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
Expand Down
157 changes: 93 additions & 64 deletions ...ache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import javax.annotation.Nullable;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
import org.apache.pinot.common.function.FunctionUtils;
Expand Down Expand Up @@ -110,13 +112,17 @@ protected enum DefaultColumnAction {
UPDATE_DIMENSION_DATA_TYPE,
UPDATE_DIMENSION_DEFAULT_VALUE,
UPDATE_DIMENSION_NUMBER_OF_VALUES,
UPDATE_DIMENSION_TRANSFORM_FUNCTION,
UPDATE_METRIC_DATA_TYPE,
UPDATE_METRIC_DEFAULT_VALUE,
UPDATE_METRIC_NUMBER_OF_VALUES,
UPDATE_METRIC_TRANSFORM_FUNCTION,
UPDATE_DATE_TIME_DATA_TYPE,
UPDATE_DATE_TIME_DEFAULT_VALUE,
UPDATE_DATE_TIME_TRANSFORM_FUNCTION,
UPDATE_COMPLEX_DATA_TYPE,
UPDATE_COMPLEX_DEFAULT_VALUE;
UPDATE_COMPLEX_DEFAULT_VALUE,
UPDATE_COMPLEX_TRANSFORM_FUNCTION;

boolean isAddAction() {
return this == ADD_DIMENSION || this == ADD_METRIC || this == ADD_DATE_TIME || this == ADD_COMPLEX;
Expand Down Expand Up @@ -293,6 +299,8 @@ Map<String, DefaultColumnAction> computeDefaultColumnActionMap() {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_DEFAULT_VALUE);
} else if (isSingleValueInMetadata != isSingleValueInSchema) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_NUMBER_OF_VALUES);
} else if (isTransformFunctionChanged(column, columnMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_TRANSFORM_FUNCTION);
}
} else if (fieldTypeInMetadata == METRIC) {
if (dataTypeInMetadata != dataTypeInSchema) {
Expand All @@ -301,18 +309,24 @@ Map<String, DefaultColumnAction> computeDefaultColumnActionMap() {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_DEFAULT_VALUE);
} else if (isSingleValueInMetadata != isSingleValueInSchema) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_NUMBER_OF_VALUES);
} else if (isTransformFunctionChanged(column, columnMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_TRANSFORM_FUNCTION);
}
} else if (fieldTypeInMetadata == DATE_TIME) {
if (dataTypeInMetadata != dataTypeInSchema) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DATE_TIME_DATA_TYPE);
} else if (!defaultValueInSchema.equals(defaultValueInMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DATE_TIME_DEFAULT_VALUE);
} else if (isTransformFunctionChanged(column, columnMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DATE_TIME_TRANSFORM_FUNCTION);
}
} else if (fieldTypeInMetadata == COMPLEX) {
if (dataTypeInMetadata != dataTypeInSchema) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_COMPLEX_DATA_TYPE);
} else if (!defaultValueInSchema.equals(defaultValueInMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_COMPLEX_DEFAULT_VALUE);
} else if (isTransformFunctionChanged(column, columnMetadata)) {
defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_COMPLEX_TRANSFORM_FUNCTION);
}
}
} else {
Expand Down Expand Up @@ -359,6 +373,11 @@ Map<String, DefaultColumnAction> computeDefaultColumnActionMap() {
return defaultColumnActionMap;
}

private boolean isTransformFunctionChanged(String column, ColumnMetadata columnMetadata) {
String transformFunctionInMetadata = columnMetadata.getTransformFunction();
return !Objects.equals(transformFunctionInMetadata, getTransformFunctionForColumn(column));
}

/**
* Helper method to update default column indices, returns {@code true} if the update succeeds, {@code false}
* otherwise.
Expand Down Expand Up @@ -390,65 +409,74 @@ protected void removeColumnIndices(String column) {
protected boolean createColumnV1Indices(String column)
throws Exception {
boolean errorOnFailure = _indexLoadingConfig.isErrorOnColumnBuildFailure();
IngestionConfig ingestionConfig = _tableConfig.getIngestionConfig();
if (ingestionConfig != null && ingestionConfig.getTransformConfigs() != null) {
List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs();
for (TransformConfig transformConfig : transformConfigs) {
if (transformConfig.getColumnName().equals(column)) {
String transformFunction = transformConfig.getTransformFunction();
FunctionEvaluator functionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);

// Check if all arguments exist in the segment
// TODO: Support chained derived column
List<String> arguments = functionEvaluator.getArguments();
List<ColumnMetadata> argumentsMetadata = new ArrayList<>(arguments.size());
for (String argument : arguments) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(argument);
if (columnMetadata == null) {
LOGGER.warn("Assigning default value to derived column: {} because argument: {} does not exist in the "
+ "segment", column, argument);
createDefaultValueColumnV1Indices(column);
return true;
}
// TODO: Support creation of derived columns from forward index disabled columns
if (!_segmentWriter.hasIndexFor(argument, StandardIndexes.forward())) {
throw new UnsupportedOperationException(String.format("Operation not supported! Cannot create a derived "
+ "column %s because argument: %s does not have a forward index. Enable forward index and "
+ "refresh/backfill the segments to create a derived column from source column", column,
argument));
}
argumentsMetadata.add(columnMetadata);
}
String transformFunction = getTransformFunctionForColumn(column);
if (transformFunction != null) {
FunctionEvaluator functionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);

// Check if all arguments exist in the segment
// TODO: Support chained derived column
List<String> arguments = functionEvaluator.getArguments();
List<ColumnMetadata> argumentsMetadata = new ArrayList<>(arguments.size());
for (String argument : arguments) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(argument);
if (columnMetadata == null) {
LOGGER.warn("Assigning default value to derived column: {} because argument: {} does not exist in the "
+ "segment", column, argument);
createDefaultValueColumnV1Indices(column, transformFunction);
return true;
}
// TODO: Support creation of derived columns from forward index disabled columns
if (!_segmentWriter.hasIndexFor(argument, StandardIndexes.forward())) {
throw new UnsupportedOperationException(String.format("Operation not supported! Cannot create a derived "
+ "column %s because argument: %s does not have a forward index. Enable forward index and "
+ "refresh/backfill the segments to create a derived column from source column", column,
argument));
}
argumentsMetadata.add(columnMetadata);
}

// TODO: Support forward index disabled derived column
if (isForwardIndexDisabled(column)) {
LOGGER.warn("Skip creating forward index disabled derived column: {}", column);
if (errorOnFailure) {
throw new UnsupportedOperationException(
String.format("Failed to create forward index disabled derived column: %s", column));
}
return false;
}
// TODO: Support forward index disabled derived column
if (isForwardIndexDisabled(column)) {
LOGGER.warn("Skip creating forward index disabled derived column: {}", column);
if (errorOnFailure) {
throw new UnsupportedOperationException(
String.format("Failed to create forward index disabled derived column: %s", column));
}
return false;
}

try {
createDerivedColumnV1Indices(column, functionEvaluator, argumentsMetadata, errorOnFailure);
return true;
} catch (Exception e) {
LOGGER.error("Caught exception while creating derived column: {} with transform function: {}", column,
transformFunction, e);
if (errorOnFailure) {
throw e;
}
return false;
}
try {
createDerivedColumnV1Indices(column, transformFunction, functionEvaluator, argumentsMetadata, errorOnFailure);
return true;
} catch (Exception e) {
LOGGER.error("Caught exception while creating derived column: {} with transform function: {}", column,
transformFunction, e);
if (errorOnFailure) {
throw e;
}
return false;
}
}

createDefaultValueColumnV1Indices(column);
createDefaultValueColumnV1Indices(column, null);
return true;
}

@SuppressWarnings("deprecation")
private String getTransformFunctionForColumn(String column) {
IngestionConfig ingestionConfig = _tableConfig.getIngestionConfig();
if (ingestionConfig != null && ingestionConfig.getTransformConfigs() != null) {
for (TransformConfig transformConfig : ingestionConfig.getTransformConfigs()) {
if (column.equals(transformConfig.getColumnName())) {
return transformConfig.getTransformFunction();
}
}
}
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);
// Keep the schema-level transform fallback for legacy configs.
return fieldSpec != null ? fieldSpec.getTransformFunction() : null;
}

/**
* Check and return whether the forward index is disabled for a given column
*/
Expand All @@ -460,7 +488,7 @@ protected boolean isForwardIndexDisabled(String column) {
/**
* Helper method to create the V1 indices (dictionary and forward index) for a column with default values.
*/
private void createDefaultValueColumnV1Indices(String column)
private void createDefaultValueColumnV1Indices(String column, @Nullable String transformFunction)
throws Exception {
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);

Expand Down Expand Up @@ -528,7 +556,7 @@ private void createDefaultValueColumnV1Indices(String column)

// Add the column metadata information to the metadata properties.
BaseSegmentCreator.addColumnMetadataInfo(_segmentProperties, column, columnStatistics, totalDocs, fieldSpec, true,
dictionaryElementSize, FieldConfig.EncodingType.DICTIONARY, true);
dictionaryElementSize, FieldConfig.EncodingType.DICTIONARY, true, transformFunction);
}

private boolean isNullable(FieldSpec fieldSpec) {
Expand All @@ -545,8 +573,8 @@ private boolean isNullable(FieldSpec fieldSpec) {
* - Support chained derived column
* - Support forward index disabled derived column
*/
private void createDerivedColumnV1Indices(String column, FunctionEvaluator functionEvaluator,
List<ColumnMetadata> argumentsMetadata, boolean errorOnFailure)
private void createDerivedColumnV1Indices(String column, String transformFunction,
FunctionEvaluator functionEvaluator, List<ColumnMetadata> argumentsMetadata, boolean errorOnFailure)
throws Exception {
// Initialize value readers for all arguments
int numArguments = argumentsMetadata.size();
Expand Down Expand Up @@ -766,10 +794,11 @@ private void createDerivedColumnV1Indices(String column, FunctionEvaluator funct
}

if (createDictionary) {
createDerivedColumnForwardIndexWithDictionary(column, fieldSpec, outputValues, statsCollector,
useVarLengthDictionary);
createDerivedColumnForwardIndexWithDictionary(column, transformFunction, fieldSpec, outputValues,
statsCollector, useVarLengthDictionary);
} else {
createDerivedColumnForwardIndexWithoutDictionary(column, fieldSpec, outputValues, statsCollector);
createDerivedColumnForwardIndexWithoutDictionary(column, transformFunction, fieldSpec, outputValues,
statsCollector);
}
} finally {
for (ValueReader valueReader : valueReaders) {
Expand Down Expand Up @@ -1083,8 +1112,8 @@ private Object getBytesOutputValue(Object outputValue, boolean isSingleValue, Pi
/**
* Helper method to create the dictionary and forward indices for a column with derived values.
*/
private void createDerivedColumnForwardIndexWithDictionary(String column, FieldSpec fieldSpec, Object[] outputValues,
ColumnStatistics columnStatistics, boolean useVarLengthDictionary)
private void createDerivedColumnForwardIndexWithDictionary(String column, String transformFunction,
FieldSpec fieldSpec, Object[] outputValues, ColumnStatistics columnStatistics, boolean useVarLengthDictionary)
throws Exception {

// Create dictionary
Expand All @@ -1111,16 +1140,16 @@ private void createDerivedColumnForwardIndexWithDictionary(String column, FieldS
}
// Add the column metadata
BaseSegmentCreator.addColumnMetadataInfo(_segmentProperties, column, columnStatistics, numDocs, fieldSpec, true,
dictionaryCreator.getNumBytesPerEntry(), FieldConfig.EncodingType.DICTIONARY, true);
dictionaryCreator.getNumBytesPerEntry(), FieldConfig.EncodingType.DICTIONARY, true, transformFunction);
}
}
}

/**
* Helper method to create a forward index for a raw encoded column with derived values.
*/
private void createDerivedColumnForwardIndexWithoutDictionary(String column, FieldSpec fieldSpec,
Object[] outputValues, ColumnStatistics columnStatistics)
private void createDerivedColumnForwardIndexWithoutDictionary(String column, String transformFunction,
FieldSpec fieldSpec, Object[] outputValues, ColumnStatistics columnStatistics)
throws Exception {

// Create forward index
Expand Down Expand Up @@ -1196,7 +1225,7 @@ private void createDerivedColumnForwardIndexWithoutDictionary(String column, Fie

// Add the column metadata
BaseSegmentCreator.addColumnMetadataInfo(_segmentProperties, column, columnStatistics, numDocs, fieldSpec, false,
0, FieldConfig.EncodingType.RAW, true);
0, FieldConfig.EncodingType.RAW, true, transformFunction);
}

private ForwardIndexCreator getForwardIndexCreator(ColumnStatistics columnStatistics,
Expand Down
6 changes: 6 additions & 0 deletions .../src/main/java/org/apache/pinot/segment/local/segment/index/map/SimpleColumnMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,12 @@ public boolean isAutoGenerated() {
return false;
}

@Nullable
@Override
public String getTransformFunction() {
return null;
}

@Override
public int getNumIndexes() {
return 0;
Expand Down
Loading