diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java index 08564230f148..ff841e06488d 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java @@ -63,6 +63,9 @@ import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNotSame; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertSame; @@ -228,6 +231,60 @@ public void testOpenStructChildStringsShared() } } + /// Every segment of a table parses the same column definitions, so the FieldSpecs are interned: two loads alias one + /// instance per column, in the column metadata and in the segment Schema alike, while the Schema object itself stays + /// per segment and the specs stay equal to the schema the segment was built from. + @Test + public void testFieldSpecsSharedAcrossLoads() + throws Exception { + SegmentMetadataImpl first = new SegmentMetadataImpl(_segmentDirectory); + SegmentMetadataImpl second = new SegmentMetadataImpl(_segmentDirectory); + assertEquals(first.getSchema(), second.getSchema()); + assertNotSame(first.getSchema(), second.getSchema()); + for (String column : first.getColumnMetadataMap().keySet()) { + FieldSpec fieldSpec = first.getColumnMetadataFor(column).getFieldSpec(); + assertSame(second.getColumnMetadataFor(column).getFieldSpec(), fieldSpec, column); + assertSame(first.getSchema().getFieldSpecFor(column), fieldSpec, column); + assertSame(second.getSchema().getFieldSpecFor(column), fieldSpec, column); + } + Schema inputSchema = SegmentTestUtils.extractSchemaFromAvroWithoutTime(_avroFile); + for (FieldSpec inputFieldSpec : inputSchema.getAllFieldSpecs()) { + assertEquals(first.getSchema().getFieldSpecFor(inputFieldSpec.getName()), inputFieldSpec); + } + + // Only the specs are shared: removing a column from one segment's schema leaves the other segment intact. + String column = first.getColumnMetadataMap().firstKey(); + first.getSchema().removeField(column); + assertNull(first.getSchema().getFieldSpecFor(column)); + assertNotNull(second.getSchema().getFieldSpecFor(column)); + assertSame(second.getSchema().getFieldSpecFor(column), second.getColumnMetadataFor(column).getFieldSpec()); + } + + /// A COMPLEX parent is not interned (ComplexFieldSpec does not override equals, so two structs with different + /// children would alias), but its children and the materialized child columns are. + @Test + public void testOpenStructChildSpecsSharedButParentIsNot() + throws Exception { + String parent = "metrics"; + File segmentDir = buildOpenStructSegment(parent); + try { + SegmentMetadataImpl first = new SegmentMetadataImpl(segmentDir); + SegmentMetadataImpl second = new SegmentMetadataImpl(segmentDir); + ComplexFieldSpec firstParent = (ComplexFieldSpec) first.getColumnMetadataFor(parent).getFieldSpec(); + ComplexFieldSpec secondParent = (ComplexFieldSpec) second.getColumnMetadataFor(parent).getFieldSpec(); + assertNotSame(secondParent, firstParent); + assertEquals(secondParent.getChildFieldSpecs(), firstParent.getChildFieldSpecs()); + for (Map.Entry entry : firstParent.getChildFieldSpecs().entrySet()) { + assertSame(secondParent.getChildFieldSpec(entry.getKey()), entry.getValue(), entry.getKey()); + } + String child = OpenStructNaming.materializedColumnName(parent, "cpu"); + assertSame(second.getColumnMetadataFor(child).getFieldSpec(), first.getColumnMetadataFor(child).getFieldSpec()); + assertSame(second.getColumnMetadataFor("dim").getFieldSpec(), first.getColumnMetadataFor("dim").getFieldSpec()); + } finally { + FileUtils.deleteQuietly(segmentDir); + } + } + private static File buildOpenStructSegment(String parent) throws Exception { Map children = new HashMap<>(); diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java index 8344cf33cb47..759eb8a02aaf 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java @@ -25,6 +25,7 @@ import org.apache.pinot.segment.spi.index.IndexType; import org.apache.pinot.spi.annotations.InterfaceAudience; import org.apache.pinot.spi.config.table.FieldConfig.EncodingType; +import org.apache.pinot.spi.data.FieldSpec; /// The `ColumnMetadata` class holds the column level management information and data statistics. @@ -32,6 +33,15 @@ public interface ColumnMetadata extends ColumnShape { int UNAVAILABLE = -1; + /// Returns the [FieldSpec] of the column. + /// + /// A spec derived from segment metadata (`metadata.properties`) is shared: every loaded segment whose column parses + /// to an equal spec, in this table or any other, holds the same instance, so it must be treated as immutable. Never + /// call a setter on it; copy it (e.g. through a JSON round-trip) before mutating. Compare specs with + /// [FieldSpec#equals], never by identity. + @Override + FieldSpec getFieldSpec(); + /// Returns `true` when the column has a dictionary, `false` otherwise. @JsonProperty("hasDictionary") boolean hasDictionary(); diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java index 462578034079..8559134b5686 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java @@ -64,6 +64,12 @@ public interface SegmentMetadata { SegmentVersion getVersion(); + /// Returns the schema of the segment, one [org.apache.pinot.spi.data.FieldSpec] per column. + /// + /// The `Schema` object itself belongs to this segment, but the specs it holds are shared with every other loaded + /// segment (of this table or any other) whose column parses to an equal spec, and must be treated as immutable: never + /// call a setter on one; copy it (e.g. through a JSON round-trip) before mutating. Removing a column from this schema + /// does not affect other segments. Schema getSchema(); int getTotalDocs(); diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java index 8ed3b65f2208..d47fa0d56b04 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java @@ -20,6 +20,8 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.Interner; +import com.google.common.collect.Interners; import com.google.common.collect.Maps; import it.unimi.dsi.fastutil.ints.IntSet; import java.math.BigDecimal; @@ -64,12 +66,22 @@ /// 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). +/// `FieldSpec.DEFAULT_*` constant and never retains the literal. The [FieldSpec] itself is then interned through +/// [#FIELD_SPEC_INTERNER], so every segment of a table (and every table with an identical column definition) shares +/// one instance per distinct spec instead of retaining its own. Segment-derived [FieldSpec]s must therefore be treated +/// as immutable: a setter call on one would bleed into every other segment and table that shares it, and would +/// corrupt the interner's hash bucket (nothing ever mutated one; copy via a JSON round-trip before mutating). @SuppressWarnings({"rawtypes", "unchecked"}) public class ColumnMetadataImpl implements ColumnMetadata { private static final long SIZE_MASK = 0xffffffffffffL; + /// Canonical instances of the [FieldSpec]s parsed from `metadata.properties`, keyed by [FieldSpec#equals] / + /// [FieldSpec#hashCode] (name, data type, single-value, default null value, max length, date-time format and + /// granularity, ...), so schema evolution yields a distinct canonical instance per version of a column. The specs + /// are held weakly: the canonical instance is exactly the one the loaded segments retain, so it lives as long as + /// any of them and is released once the last one is unloaded. Thread-safe. + private static final Interner FIELD_SPEC_INTERNER = Interners.newWeakInterner(); + private final FieldSpec _fieldSpec; private final int _totalDocs; private final int _cardinality; @@ -504,6 +516,11 @@ private static ChunkCompressionType parseCompressionType(String column, @Nullabl } } + /// Parses the [FieldSpec] of the given column. DIMENSION, METRIC, TIME and DATE_TIME specs are returned from + /// [#FIELD_SPEC_INTERNER], so the instance is shared with every other segment whose column parses to an equal spec + /// and must not be mutated. A COMPLEX spec is not interned: [ComplexFieldSpec] does not override + /// [FieldSpec#equals], so two structs with different children would alias; its children are parsed through this + /// method and are interned. public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration config) { // 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 @@ -524,19 +541,20 @@ public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration ? FieldSpec.MaxLengthExceedStrategy.valueOf(maxLengthExceedStrategyString) : null; switch (fieldType) { case DIMENSION: - return new DimensionFieldSpec(fieldName, dataType, isSingleValue, maxLength, - canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), maxLengthExceedStrategy); + return FIELD_SPEC_INTERNER.intern(new DimensionFieldSpec(fieldName, dataType, isSingleValue, maxLength, + canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), maxLengthExceedStrategy)); case METRIC: - return new MetricFieldSpec(fieldName, dataType, - canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), maxLength, maxLengthExceedStrategy); + return FIELD_SPEC_INTERNER.intern(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)); + return FIELD_SPEC_INTERNER.intern(new TimeFieldSpec(new TimeGranularitySpec(dataType, timeUnit, fieldName))); case DATE_TIME: 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); + return FIELD_SPEC_INTERNER.intern(new DateTimeFieldSpec(fieldName, dataType, format, granularity, + canonicalDefaultNullValue(fieldType, dataType, defaultNullValueString), null)); case COMPLEX: List childFieldNames = config.getList(String.class, Column.getKeyFor(column, Column.COMPLEX_CHILD_FIELD_NAMES)); @@ -547,6 +565,7 @@ public static FieldSpec extractFieldSpec(String column, PropertiesConfiguration extractFieldSpec(ComplexFieldSpec.getFullChildName(column, childField), config)); } } + // Deliberately not interned (see the method doc): only the children above are shared. return new ComplexFieldSpec(fieldName, dataType, true, childFieldSpecs); default: throw new IllegalStateException("Unsupported field type: " + fieldType); diff --git a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImplTest.java b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImplTest.java index 44913ad14e95..6e6aea4e7b93 100644 --- a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImplTest.java +++ b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImplTest.java @@ -19,20 +19,29 @@ package org.apache.pinot.segment.spi.index.metadata; import com.fasterxml.jackson.databind.JsonNode; +import java.lang.ref.WeakReference; import java.math.BigDecimal; +import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; import org.apache.commons.configuration2.PropertiesConfiguration; import org.apache.pinot.segment.spi.ColumnMetadata; import org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column; +import org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Segment; import org.apache.pinot.segment.spi.compression.ChunkCompressionType; import org.apache.pinot.spi.config.table.FieldConfig.EncodingType; +import org.apache.pinot.spi.data.ComplexFieldSpec; import org.apache.pinot.spi.data.DateTimeFieldSpec; 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.FieldSpec.FieldType; import org.apache.pinot.spi.data.MetricFieldSpec; +import org.apache.pinot.spi.data.TimeFieldSpec; +import org.apache.pinot.spi.data.TimeGranularitySpec; import org.apache.pinot.spi.env.CommonsConfigurationUtils; import org.apache.pinot.spi.utils.BytesUtils; import org.apache.pinot.spi.utils.JsonUtils; @@ -441,6 +450,132 @@ public void columnNameAndParentColumnAreInterned() { assertSame(spec.getName(), "plain"); } + /// A server retains one FieldSpec per (segment, column) and every segment of a table parses the same column + /// definition, so the parse path interns the spec: two loads of the same metadata, or of two segments whose column + /// parses to an equal spec, alias one instance that is still equal to the spec the table schema builds. + @Test + public void equalSpecsAreInterned() { + PropertiesConfiguration config = baseConfig("col"); + FieldSpec spec = ColumnMetadataImpl.fromPropertiesConfiguration(config, 1, "col").getFieldSpec(); + assertSame(ColumnMetadataImpl.fromPropertiesConfiguration(config, 2, "col").getFieldSpec(), spec); + assertSame(ColumnMetadataImpl.fromPropertiesConfiguration(baseConfig("col"), 3, "col").getFieldSpec(), spec); + assertEquals(spec, new DimensionFieldSpec("col", DataType.STRING, true)); + + // A custom default and a max length are part of the key and are shared as well. + FieldSpec custom = parse(FieldType.DIMENSION, DataType.INT, "-1"); + assertSame(parse(FieldType.DIMENSION, DataType.INT, "-1"), custom); + assertEquals(custom, new DimensionFieldSpec("col", DataType.INT, true, -1)); + PropertiesConfiguration bounded = configFor(FieldType.DIMENSION, DataType.STRING, null); + bounded.setProperty(Column.getKeyFor("col", Column.SCHEMA_MAX_LENGTH), 10); + FieldSpec boundedSpec = ColumnMetadataImpl.extractFieldSpec("col", bounded); + assertSame(ColumnMetadataImpl.extractFieldSpec("col", bounded), boundedSpec); + assertEquals(boundedSpec, new DimensionFieldSpec("col", DataType.STRING, true, 10, null)); + } + + @Test + public void metricTimeAndDateTimeSpecsAreInterned() { + FieldSpec metric = parse(FieldType.METRIC, DataType.LONG, null); + assertSame(parse(FieldType.METRIC, DataType.LONG, null), metric); + assertEquals(metric, new MetricFieldSpec("col", DataType.LONG)); + + PropertiesConfiguration hours = configFor(FieldType.TIME, DataType.INT, null); + hours.setProperty(Segment.TIME_UNIT, "HOURS"); + FieldSpec time = ColumnMetadataImpl.extractFieldSpec("col", hours); + assertSame(ColumnMetadataImpl.extractFieldSpec("col", hours), time); + assertEquals(time, new TimeFieldSpec(new TimeGranularitySpec(DataType.INT, TimeUnit.HOURS, "col"))); + // The time unit is part of the TimeFieldSpec key. + PropertiesConfiguration days = configFor(FieldType.TIME, DataType.INT, null); + days.setProperty(Segment.TIME_UNIT, "DAYS"); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("col", days), time); + + FieldSpec dateTime = parse(FieldType.DATE_TIME, DataType.LONG, null); + assertSame(parse(FieldType.DATE_TIME, DataType.LONG, null), dateTime); + assertEquals(dateTime, new DateTimeFieldSpec("col", DataType.LONG, DATETIME_FORMAT, DATETIME_GRANULARITY)); + } + + /// Interning is keyed by [FieldSpec#equals], so a column whose definition changed (schema evolution) parses to a + /// distinct canonical instance instead of aliasing the previous one. + @Test + public void differingSpecsAreNotInterned() { + FieldSpec base = parse(FieldType.DIMENSION, DataType.INT, null); + assertNotSame(parse(FieldType.DIMENSION, DataType.INT, "-1"), base, "default null value"); + assertNotSame(parse(FieldType.DIMENSION, DataType.LONG, null), base, "data type"); + assertNotSame(parse(FieldType.METRIC, DataType.INT, null), base, "field type"); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("other", baseConfig("other")), + ColumnMetadataImpl.extractFieldSpec("col", baseConfig("col")), "name"); + PropertiesConfiguration multiValue = configFor(FieldType.DIMENSION, DataType.INT, null); + multiValue.setProperty(Column.getKeyFor("col", Column.IS_SINGLE_VALUED), false); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("col", multiValue), base, "single value"); + PropertiesConfiguration maxLength = configFor(FieldType.DIMENSION, DataType.INT, null); + maxLength.setProperty(Column.getKeyFor("col", Column.SCHEMA_MAX_LENGTH), 10); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("col", maxLength), base, "max length"); + + FieldSpec dateTime = parse(FieldType.DATE_TIME, DataType.LONG, null); + PropertiesConfiguration otherFormat = configFor(FieldType.DATE_TIME, DataType.LONG, null); + otherFormat.setProperty(Column.getKeyFor("col", Column.DATETIME_FORMAT), "1:SECONDS:EPOCH"); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("col", otherFormat), dateTime, "format"); + PropertiesConfiguration otherGranularity = configFor(FieldType.DATE_TIME, DataType.LONG, null); + otherGranularity.setProperty(Column.getKeyFor("col", Column.DATETIME_GRANULARITY), "1:SECONDS"); + assertNotSame(ColumnMetadataImpl.extractFieldSpec("col", otherGranularity), dateTime, "granularity"); + } + + /// [ComplexFieldSpec] does not override equals/hashCode, so two structs with the same name but different children + /// are equal under [FieldSpec#equals]; interning the parent would alias them. Only the children are interned. + @Test + public void complexParentIsNotInternedWhileChildrenAre() { + PropertiesConfiguration twoChildren = complexConfig("metrics", "cpu", "host"); + ComplexFieldSpec first = (ComplexFieldSpec) ColumnMetadataImpl.extractFieldSpec("metrics", twoChildren); + ComplexFieldSpec second = (ComplexFieldSpec) ColumnMetadataImpl.extractFieldSpec("metrics", twoChildren); + ComplexFieldSpec narrower = + (ComplexFieldSpec) ColumnMetadataImpl.extractFieldSpec("metrics", complexConfig("metrics", "cpu")); + assertNotSame(second, first); + assertNotSame(narrower, first); + // The guard is real: the parents are equal despite their different children. + assertEquals(narrower, first); + assertEquals(first.getChildFieldSpecs().keySet(), Set.of("cpu", "host")); + assertEquals(narrower.getChildFieldSpecs().keySet(), Set.of("cpu")); + assertSame(second.getChildFieldSpec("cpu"), first.getChildFieldSpec("cpu")); + assertSame(second.getChildFieldSpec("host"), first.getChildFieldSpec("host")); + assertSame(narrower.getChildFieldSpec("cpu"), first.getChildFieldSpec("cpu")); + assertEquals(first.getChildFieldSpec("cpu"), + new DimensionFieldSpec(ComplexFieldSpec.getFullChildName("metrics", "cpu"), DataType.DOUBLE, true)); + } + + /// The interner holds its specs weakly, so a spec is released once the last segment referencing it is unloaded; an + /// interner that pinned them would leak one spec per column definition ever loaded. + @Test + public void unreferencedSpecIsReleased() + throws InterruptedException { + WeakReference spec = internUnreferencedSpec(); + for (int i = 0; i < 100 && spec.get() != null; i++) { + System.gc(); + Thread.sleep(10); + } + assertNull(spec.get(), "the interner must not keep an unloaded segment's spec alive"); + } + + /// Parses a spec no other test builds (a random custom default) and hands back only a weak reference to it. + private static WeakReference internUnreferencedSpec() { + return new WeakReference<>(parse(FieldType.DIMENSION, DataType.STRING, "unreferenced-" + UUID.randomUUID())); + } + + /// A COMPLEX parent with DOUBLE children, written the way the segment creator writes it. + private static PropertiesConfiguration complexConfig(String parent, String... children) { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(Column.getKeyFor(parent, Column.COLUMN_NAME), parent); + config.setProperty(Column.getKeyFor(parent, Column.COLUMN_TYPE), FieldType.COMPLEX.name()); + config.setProperty(Column.getKeyFor(parent, Column.DATA_TYPE), DataType.OPEN_STRUCT.name()); + config.setProperty(Column.getKeyFor(parent, Column.IS_SINGLE_VALUED), true); + config.setProperty(Column.getKeyFor(parent, Column.COMPLEX_CHILD_FIELD_NAMES), List.of(children)); + for (String child : children) { + String column = ComplexFieldSpec.getFullChildName(parent, child); + config.setProperty(Column.getKeyFor(column, Column.COLUMN_TYPE), FieldType.DIMENSION.name()); + config.setProperty(Column.getKeyFor(column, Column.DATA_TYPE), DataType.DOUBLE.name()); + config.setProperty(Column.getKeyFor(column, Column.IS_SINGLE_VALUED), true); + } + return config; + } + private static FieldSpec parse(FieldType fieldType, DataType dataType, @Nullable String defaultNullValue) { return ColumnMetadataImpl.extractFieldSpec("col", configFor(fieldType, dataType, defaultNullValue)); }