Skip to content

Canonicalize the default null value and intern per-column strings at metadata parse time #19473

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
xiangfu0 wants to merge 1 commit into from
Open

Canonicalize the default null value and intern per-column strings at metadata parse time #19473

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
124 changes: 123 additions & 1 deletion ...l/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,23 @@
package org.apache.pinot.segment.local.segment.index;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.converter.SegmentV1V2ToV3FormatConverter;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
Expand All @@ -37,29 +44,44 @@
import org.apache.pinot.segment.spi.index.metadata.ColumnMetadataImpl;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.OpenStructNaming;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertSame;


public class SegmentMetadataImplTest {
private static final String AVRO_DATA = "data/test_data-mv.avro";
private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(), "SegmentMetadataImplTest");
private File _avroFile;
private File _segmentDirectory;

@BeforeMethod
public void setUp()
throws Exception {
final String filePath =
TestUtils.getFileFromResourceUrl(SegmentMetadataImplTest.class.getClassLoader().getResource(AVRO_DATA));
_avroFile = new File(filePath);

// intentionally changed this to TimeUnit.Hours to make it non-default for testing
final SegmentGeneratorConfig config = SegmentTestUtils
.getSegmentGenSpecWithSchemAndProjectedColumns(new File(filePath), INDEX_DIR, "daysSinceEpoch", TimeUnit.HOURS,
.getSegmentGenSpecWithSchemAndProjectedColumns(_avroFile, INDEX_DIR, "daysSinceEpoch", TimeUnit.HOURS,
"testTable");
config.setSegmentNamePostfix("1");
config.setCustomProperties(Map.of("custom.k1", "v1", "custom.k2", "v2"));
Expand Down Expand Up @@ -141,4 +163,104 @@ public void testIndexSizesOnlyFromIndexDir()
Assert.assertTrue(((ColumnMetadataImpl) streamColumn).getIndexSizeMap().isEmpty(), entry.getKey());
}
}

/// A server holds one column-metadata graph per loaded segment, so the per-column strings and default null values
/// are shared: two loads of the same metadata alias one interned column name (as the map key, the FieldSpec name and
/// the Schema entry) and hold the static FieldSpec default constant rather than a box parsed from the literal the
/// segment creator wrote, while the specs stay equal to the schema the segment was built from.
@Test
public void testColumnStringsAndDefaultsSharedAcrossLoads()
throws Exception {
SegmentMetadataImpl first = new SegmentMetadataImpl(_segmentDirectory);
SegmentMetadataImpl second = new SegmentMetadataImpl(_segmentDirectory);
assertEquals(first.getColumnMetadataMap().keySet(), second.getColumnMetadataMap().keySet());
assertSame(first.getColumnMetadataMap().firstKey(), second.getColumnMetadataMap().firstKey());
assertSame(first.getSchema().getDimensionNames().get(0), second.getSchema().getDimensionNames().get(0));
Iterator<String> secondKeys = second.getColumnMetadataMap().keySet().iterator();
for (String column : first.getColumnMetadataMap().keySet()) {
assertSame(column, secondKeys.next());
FieldSpec fieldSpec = first.getColumnMetadataFor(column).getFieldSpec();
assertSame(fieldSpec.getName(), column, column);
assertSame(fieldSpec.getName(), second.getColumnMetadataFor(column).getFieldSpec().getName(), column);
assertSame(first.getSchema().getFieldSpecFor(column).getName(), column, column);
assertSame(fieldSpec.getDefaultNullValue(),
FieldSpec.getDefaultNullValue(fieldSpec.getFieldType(), fieldSpec.getDataType(), null), column);
}
Schema inputSchema = SegmentTestUtils.extractSchemaFromAvroWithoutTime(_avroFile);
for (FieldSpec inputFieldSpec : inputSchema.getAllFieldSpecs()) {
assertEquals(first.getSchema().getFieldSpecFor(inputFieldSpec.getName()), inputFieldSpec);
}
}

/// OPEN_STRUCT children carry an explicit column name and a parent column in `metadata.properties`; both come back
/// as the interned instances, so the child's parent name is the very String that keys the parent column.
@Test
public void testOpenStructChildStringsShared()
throws Exception {
String parent = "metrics";
File segmentDir = buildOpenStructSegment(parent);
try {
SegmentMetadataImpl first = new SegmentMetadataImpl(segmentDir);
SegmentMetadataImpl second = new SegmentMetadataImpl(segmentDir);
String parentKey = first.getColumnMetadataMap().ceilingKey(parent);
assertEquals(parentKey, parent);
assertSame(parentKey, second.getColumnMetadataMap().ceilingKey(parent));
String child = OpenStructNaming.materializedColumnName(parent, "cpu");
ColumnMetadataImpl firstChild = (ColumnMetadataImpl) first.getColumnMetadataFor(child);
ColumnMetadataImpl secondChild = (ColumnMetadataImpl) second.getColumnMetadataFor(child);
assertEquals(firstChild.getParentColumn(), parent);
assertSame(firstChild.getParentColumn(), parentKey);
assertSame(firstChild.getParentColumn(), secondChild.getParentColumn());
assertSame(firstChild.getFieldSpec().getName(), secondChild.getFieldSpec().getName());
assertSame(firstChild.getFieldSpec().getDefaultNullValue(), FieldSpec.DEFAULT_DIMENSION_NULL_VALUE_OF_DOUBLE);
ComplexFieldSpec firstParent = (ComplexFieldSpec) first.getColumnMetadataFor(parent).getFieldSpec();
ComplexFieldSpec secondParent = (ComplexFieldSpec) second.getColumnMetadataFor(parent).getFieldSpec();
assertEquals(firstParent.getChildFieldSpecs().keySet(), Set.of("views", "cpu", "host"));
for (Map.Entry<String, FieldSpec> entry : firstParent.getChildFieldSpecs().entrySet()) {
FieldSpec childSpec = entry.getValue();
assertSame(childSpec.getName(), entry.getKey());
assertSame(childSpec.getName(), secondParent.getChildFieldSpec(entry.getKey()).getName());
assertSame(childSpec.getDefaultNullValue(),
FieldSpec.getDefaultNullValue(childSpec.getFieldType(), childSpec.getDataType(), null));
}
} finally {
FileUtils.deleteQuietly(segmentDir);
}
}

private static File buildOpenStructSegment(String parent)
throws Exception {
Map<String, FieldSpec> children = new HashMap<>();
children.put("views", new DimensionFieldSpec("views", FieldSpec.DataType.LONG, true));
children.put("cpu", new DimensionFieldSpec("cpu", FieldSpec.DataType.DOUBLE, true));
children.put("host", new DimensionFieldSpec("host", FieldSpec.DataType.STRING, true));
Schema schema = new Schema.SchemaBuilder().setSchemaName("testOpenStruct")
.addField(new ComplexFieldSpec(parent, FieldSpec.DataType.OPEN_STRUCT, true, children))
.addSingleValueDimension("dim", FieldSpec.DataType.STRING)
.build();
OpenStructIndexConfig openStructConfig =
new OpenStructIndexConfig(false, null, 3, Set.of("views", "cpu", "host"), 0.5, List.of(), null);
ObjectNode indexes = JsonUtils.newObjectNode();
indexes.set("open_struct", JsonUtils.objectToJsonNode(openStructConfig));
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testOpenStruct")
.setFieldConfigList(List.of(new FieldConfig.Builder(parent).withIndexes(indexes).build())).build();
SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, schema);
config.setOutDir(new File(INDEX_DIR, "openStruct").getAbsolutePath());
config.setSegmentName("openStructSegment");
List<GenericRow> rows = new ArrayList<>();
for (int i = 0; i < 10; i++) {
GenericRow row = new GenericRow();
Map<String, Object> metrics = new HashMap<>();
metrics.put("views", (long) i);
metrics.put("cpu", i * 0.5);
metrics.put("host", "host-" + i);
row.putValue(parent, metrics);
row.putValue("dim", "val-" + i);
rows.add(row);
}
SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
driver.init(config, new GenericRowRecordReader(rows));
driver.build();
return driver.getOutputDirectory();
}
}
64 changes: 55 additions & 9 deletions ...ent-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.pinot.segment.spi.index.metadata;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import it.unimi.dsi.fastutil.ints.IntSet;
import java.math.BigDecimal;
Expand Down Expand Up @@ -57,6 +58,14 @@
import org.apache.pinot.spi.utils.JsonUtils;


/// Column metadata parsed from `metadata.properties` (or built through [Builder]).
///
/// A server retains one instance per (segment, column) for as long as the segment is loaded, so the parse path keeps
/// the per-column footprint small: column names, parent-column names, date-time formats/granularities and custom
/// default-null literals are interned (they recur in every segment of a table), and a `defaultNullValue` that equals
/// the type default is not handed to the [FieldSpec] at all, so the spec carries the shared static
/// `FieldSpec.DEFAULT_*` constant and never retains the literal. Segment-derived [FieldSpec]s must therefore be
/// treated as read-only: nothing may mutate their default null value in place (nothing ever did).
@SuppressWarnings({"rawtypes", "unchecked"})
public class ColumnMetadataImpl implements ColumnMetadata {
private static final long SIZE_MASK = 0xffffffffffffL;
Expand Down Expand Up @@ -419,7 +428,7 @@ public static ColumnMetadataImpl fromPropertiesConfiguration(PropertiesConfigura
.setMaxRowLengthInBytes(config.getInt(Column.getKeyFor(column, Column.MAX_ROW_LENGTH_IN_BYTES), UNAVAILABLE))
.setBitsPerElement(config.getInt(Column.getKeyFor(column, Column.BITS_PER_ELEMENT), UNAVAILABLE))
.setAutoGenerated(config.getBoolean(Column.getKeyFor(column, Column.IS_AUTO_GENERATED), false))
.setParentColumn(config.getString(Column.getKeyFor(column, Column.PARENT_COLUMN), null));
.setParentColumn(intern(config.getString(Column.getKeyFor(column, Column.PARENT_COLUMN), null)));

Object rawSparseKeys = config.getProperty(Column.getKeyFor(column, Column.SPARSE_KEYS));
if (rawSparseKeys != null) {
Expand Down Expand Up @@ -496,7 +505,11 @@ private static ChunkCompressionType parseCompressionType(String column, @Nullabl
}

public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration config) {
String fieldName = config.getString(Column.getKeyFor(column, Column.COLUMN_NAME), column);
// The name is retained by the FieldSpec, the segment Schema and every per-segment column map, and it recurs in
// every segment of the table: intern it so all of them alias one JVM-wide instance. When COLUMN_NAME is absent
// (the segment creator only writes it when it differs from the key) this is the key parsed by SegmentMetadataImpl,
// which is already interned, so the lookup just returns it.
String fieldName = config.getString(Column.getKeyFor(column, Column.COLUMN_NAME), column).intern();
FieldType fieldType = config.getEnum(Column.getKeyFor(column, Column.COLUMN_TYPE), FieldType.class);
DataType dataType = config.getEnum(Column.getKeyFor(column, Column.DATA_TYPE), DataType.class);
boolean isSingleValue = config.getBoolean(Column.getKeyFor(column, Column.IS_SINGLE_VALUED), true);
Expand All @@ -511,24 +524,26 @@ public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration
? FieldSpec.MaxLengthExceedStrategy.valueOf(maxLengthExceedStrategyString) : null;
switch (fieldType) {
case DIMENSION:
return new DimensionFieldSpec(fieldName, dataType, isSingleValue, maxLength, defaultNullValueString,
maxLengthExceedStrategy);
return new DimensionFieldSpec(fieldName, dataType, isSingleValue, maxLength,
canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), maxLengthExceedStrategy);
case METRIC:
return new MetricFieldSpec(fieldName, dataType, defaultNullValueString, maxLength, maxLengthExceedStrategy);
return new MetricFieldSpec(fieldName, dataType,
canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), maxLength, maxLengthExceedStrategy);
case TIME:
TimeUnit timeUnit = TimeUnit.valueOf(config.getString(Segment.TIME_UNIT, "DAYS").toUpperCase());
return new TimeFieldSpec(new TimeGranularitySpec(dataType, timeUnit, fieldName));
case DATE_TIME:
String format = config.getString(Column.getKeyFor(column, Column.DATETIME_FORMAT));
String granularity = config.getString(Column.getKeyFor(column, Column.DATETIME_GRANULARITY));
return new DateTimeFieldSpec(fieldName, dataType, format, granularity, defaultNullValueString, null);
String format = intern(config.getString(Column.getKeyFor(column, Column.DATETIME_FORMAT)));
String granularity = intern(config.getString(Column.getKeyFor(column, Column.DATETIME_GRANULARITY)));
return new DateTimeFieldSpec(fieldName, dataType, format, granularity,
canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), null);
case COMPLEX:
List<String> childFieldNames =
config.getList(String.class, Column.getKeyFor(column, Column.COMPLEX_CHILD_FIELD_NAMES));
Map<String, FieldSpec> childFieldSpecs = new HashMap<>();
if (childFieldNames != null) {
for (String childField : childFieldNames) {
childFieldSpecs.put(childField,
childFieldSpecs.put(childField.intern(),
extractFieldSpec(ComplexFieldSpec.getFullChildName(column, childField), config));
}
}
Expand All @@ -538,6 +553,37 @@ public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration
}
}

/// Returns the `defaultNullValue` literal to hand to the [FieldSpec] constructor: `null` when the literal parses to
/// the type default, so the spec ends up holding the shared static `FieldSpec.DEFAULT_*` constant instead of a
/// per-segment box plus the literal (the segment creator writes the literal for every column, so without this every
/// column of every segment paid for it); otherwise the interned literal, so a custom default is shared across the
/// segments of the table. Equality is [DataType#equals(Object, Object)], the predicate [FieldSpec#equals] applies to
/// default null values, so the canonical spec equals one built from the literal and
/// [FieldSpec#getDefaultNullValueString()] (derived from the value) is unchanged; a BIG_DECIMAL literal with a
/// different scale or a negative-zero FLOAT/DOUBLE is not equal and stays verbatim.
@VisibleForTesting
@Nullable
static String canonicalDefaultNullValue(FieldType fieldType, DataType dataType, @Nullable String literal) {
if (literal == null) {
return null;
}
Object typeDefault;
try {
typeDefault = FieldSpec.getDefaultNullValue(fieldType, dataType, null);
} catch (IllegalStateException e) {
// No type default for this combination (e.g. a METRIC BOOLEAN): the literal is the only valid value, exactly as
// the FieldSpec constructor treats it.
return literal.intern();
}
return dataType.equals(FieldSpec.getDefaultNullValue(fieldType, dataType, literal), typeDefault) ? null
: literal.intern();
}

@Nullable
private static String intern(@Nullable String value) {
return value != null ? value.intern() : null;
}

@Nullable
public static PartitionFunction extractPartitionFunction(String column, PropertiesConfiguration config) {
String partitionFunctionName = config.getString(Column.getKeyFor(column, Column.PARTITION_FUNCTION), null);
Expand Down
7 changes: 6 additions & 1 deletion ...nt-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,9 +311,14 @@ private static void setCustomConfigs(Configuration segmentMetadataPropertiesConf
}

/// Helper method to add the physical columns from source list to destination set.
///
/// Column names are interned: the same names recur in every segment of a table and each one is retained by the
/// column metadata map key, the FieldSpec, the segment Schema and the loader's per-column maps, so one JVM-wide
/// instance replaces a copy per segment (the JVM string table holds them weakly, so they live exactly as long as a
/// loaded segment references them).
private static void addPhysicalColumns(List<Object> src, Set<String> dest) {
for (Object o : src) {
String column = o.toString();
String column = o.toString().intern();
if (!column.isEmpty() && !BuiltInVirtualColumn.BUILT_IN_VIRTUAL_COLUMNS.contains(column)) {
// NOTE:
// Exclude built in virtual columns. In regular case they shouldn't exist in the metadata file, but we perform
Expand Down
Loading
Loading