You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/09/16 09:30:40 UTC
(pinot) 01/01: Reuse table schema field specs after segment preprocessing
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex/schema-fieldspec-reuse
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit efef2d3fa6021ce95164a83b0ff9d5c287fc7936
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Wed Sep 16 02:30:16 2026 -0700
Reuse table schema field specs after segment preprocessing
---
.../core/data/manager/BaseTableDataManager.java | 6 +-
.../pinot/core/data/manager/TableSchemaCache.java | 62 +++++
.../core/data/manager/TableSchemaCacheTest.java | 241 ++++++++++++++++++++
.../immutable/ImmutableSegmentLoader.java | 2 +
.../segment/index/loader/IndexLoadingConfig.java | 19 +-
.../index/loader/SchemaFieldSpecReuseTest.java | 215 ++++++++++++++++++
.../spi/index/metadata/ColumnMetadataImpl.java | 12 +-
.../spi/index/metadata/SegmentMetadataImpl.java | 56 +++++
.../SegmentMetadataFieldSpecReuseTest.java | 252 +++++++++++++++++++++
9 files changed, 855 insertions(+), 10 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
index 4da4cf0d6ef..7089b203400 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
@@ -135,6 +135,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected final ConcurrentHashMap<String, SegmentDataManager> _segmentDataManagerMap = new ConcurrentHashMap<>();
protected final ServerMetrics _serverMetrics = ServerMetrics.get();
+ private final TableSchemaCache _tableSchemaCache = new TableSchemaCache();
protected TableUpsertMetadataManager _tableUpsertMetadataManager;
protected InstanceDataManagerConfig _instanceDataManagerConfig;
@@ -431,9 +432,10 @@ public abstract class BaseTableDataManager implements TableDataManager {
Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", _tableNameWithType);
Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
Preconditions.checkState(schema != null, "Failed to find schema for table: %s", _tableNameWithType);
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
+ IndexLoadingConfig indexLoadingConfig =
+ new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema, _tableSchemaCache::canonicalize);
indexLoadingConfig.setTableDataDir(_tableDataDir);
- updateCachedTableConfigAndSchema(tableConfig, schema);
+ updateCachedTableConfigAndSchema(tableConfig, indexLoadingConfig.getSchema());
return indexLoadingConfig;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/TableSchemaCache.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/TableSchemaCache.java
new file mode 100644
index 00000000000..16165b2ef01
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/TableSchemaCache.java
@@ -0,0 +1,62 @@
+/**
+ * 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.data.manager;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.ComplexFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+
+
+/// Shares the latest normalized schema across a table's segment loads. Concurrent equal inputs reuse one instance;
+/// a changed schema replaces the cached instance without modifying previously returned schemas. Callers must treat
+/// returned schemas as read-only. Only the latest schema is retained by this cache.
+final class TableSchemaCache {
+ @Nullable
+ private Schema _schema;
+
+ synchronized Schema canonicalize(Schema schema) {
+ if (_schema != null && _schema.equals(schema)
+ && equalFieldSpecs(_schema.getFieldSpecMap(), schema.getFieldSpecMap())) {
+ return _schema;
+ }
+ _schema = schema;
+ return schema;
+ }
+
+ // ComplexFieldSpec.equals() does not compare its children. Check them recursively before sharing a schema.
+ private static boolean equalFieldSpecs(Map<String, FieldSpec> left, Map<String, FieldSpec> right) {
+ if (!left.keySet().equals(right.keySet())) {
+ return false;
+ }
+ for (Map.Entry<String, FieldSpec> entry : left.entrySet()) {
+ FieldSpec leftSpec = entry.getValue();
+ FieldSpec rightSpec = right.get(entry.getKey());
+ if (!leftSpec.equals(rightSpec)) {
+ return false;
+ }
+ if (leftSpec instanceof ComplexFieldSpec && !equalFieldSpecs(
+ ((ComplexFieldSpec) leftSpec).getChildFieldSpecs(), ((ComplexFieldSpec) rightSpec).getChildFieldSpecs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableSchemaCacheTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableSchemaCacheTest.java
new file mode 100644
index 00000000000..fc3f5b1454f
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableSchemaCacheTest.java
@@ -0,0 +1,241 @@
+/**
+ * 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.data.manager;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.utils.helix.FakePropertyStore;
+import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.TimestampConfig;
+import org.apache.pinot.spi.config.table.TimestampIndexGranularity;
+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.Schema;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies table-scoped schema identity across fresh loads, concurrent loads, and schema evolution.
+public class TableSchemaCacheTest {
+ @Test
+ public void testFreshTableManagerFetchesShareSchemaAndObserveChanges() {
+ BaseTableDataManager manager = new OfflineTableDataManager();
+ manager._propertyStore = new FakePropertyStore();
+ manager._tableNameWithType = "testTable_OFFLINE";
+ TableConfig table = timestampTable(TimestampIndexGranularity.DAY);
+ ZKMetadataProvider.setTableConfig(manager._propertyStore, table);
+ ZKMetadataProvider.setSchema(manager._propertyStore, schema());
+
+ IndexLoadingConfig first = manager.fetchIndexLoadingConfig();
+ IndexLoadingConfig second = manager.fetchIndexLoadingConfig();
+ assertNotSame(first.getTableConfig(), second.getTableConfig());
+ assertSame(second.getSchema(), first.getSchema());
+ assertSame(second.getSchema().getFieldSpecFor("id"), first.getSchema().getFieldSpecFor("id"));
+ assertSame(second.getSchema().getFieldSpecFor("$ts$DAY"), first.getSchema().getFieldSpecFor("$ts$DAY"));
+ assertSame(manager.getCachedTableConfigAndSchema().getRight(), second.getSchema());
+
+ Schema changed = schema();
+ changed.getFieldSpecFor("id").setDefaultNullValue(-2);
+ ZKMetadataProvider.setSchema(manager._propertyStore, changed);
+ IndexLoadingConfig third = manager.fetchIndexLoadingConfig();
+ assertNotSame(third.getSchema(), first.getSchema());
+ assertEquals(third.getSchema().getFieldSpecFor("id").getDefaultNullValue(), -2);
+ assertEquals(first.getSchema().getFieldSpecFor("id").getDefaultNullValue(), -1);
+ assertSame(manager.getCachedTableConfigAndSchema().getRight(), third.getSchema());
+ assertSame(manager.fetchIndexLoadingConfig().getSchema(), third.getSchema());
+ }
+
+ @Test
+ public void testEqualFreshSchemasReuseLatestInstance()
+ throws Exception {
+ TableSchemaCache cache = new TableSchemaCache();
+ Schema first = schema();
+ Schema fresh = Schema.fromString(first.toSingleLineJsonString());
+ assertNotSame(first, fresh);
+ assertSame(cache.canonicalize(first), first);
+ assertSame(cache.canonicalize(fresh), first);
+
+ // Separate table managers must not share their schemas through this cache.
+ assertSame(new TableSchemaCache().canonicalize(fresh), fresh);
+ }
+
+ @DataProvider
+ public Object[][] schemaChanges() {
+ return new Object[][]{
+ {(Consumer<Schema>) schema -> schema.getFieldSpecFor("id").setDefaultNullValue(-2)},
+ {(Consumer<Schema>) schema -> schema.getFieldSpecFor("id").setDataType(DataType.LONG)},
+ {(Consumer<Schema>) schema -> schema.getFieldSpecFor("id").setNotNull(true)},
+ {(Consumer<Schema>) schema -> schema.setEnableColumnBasedNullHandling(true)},
+ {(Consumer<Schema>) schema -> schema.getFieldSpecFor("id").setDescription("changed")},
+ {(Consumer<Schema>) schema -> schema.setPrimaryKeyColumns(List.of("id"))}
+ };
+ }
+
+ @Test(dataProvider = "schemaChanges")
+ public void testSchemaChangesReplaceLatestInstance(Consumer<Schema> change) {
+ TableSchemaCache cache = new TableSchemaCache();
+ Schema first = schema();
+ cache.canonicalize(first);
+ Schema changed = schema();
+ change.accept(changed);
+ assertSame(cache.canonicalize(changed), changed);
+ assertEquals(first.getFieldSpecFor("id").getDefaultNullValue(), -1);
+ assertFalse(first.isEnableColumnBasedNullHandling());
+
+ // An old version is not retained in a history map after a different version replaces it.
+ Schema reverted = schema();
+ assertSame(cache.canonicalize(reverted), reverted);
+ }
+
+ @Test
+ public void testChangedSampleValueIsNotHiddenByJsonSerialization() {
+ TableSchemaCache cache = new TableSchemaCache();
+ Schema first = schema();
+ cache.canonicalize(first);
+ Schema changed = schema();
+ ((DateTimeFieldSpec) changed.getFieldSpecFor("ts")).setSampleValue("1000");
+ assertEquals(changed.toJsonObject(), first.toJsonObject());
+ assertSame(cache.canonicalize(changed), changed);
+ }
+
+ @Test
+ public void testNestedComplexChangesAreNotHiddenBySchemaEquality() {
+ TableSchemaCache cache = new TableSchemaCache();
+ Schema first = complexSchema();
+ cache.canonicalize(first);
+ assertSame(cache.canonicalize(complexSchema()), first);
+
+ Schema changedDefault = complexSchema();
+ nestedValue(changedDefault).setDefaultNullValue(-2);
+ assertEquals(changedDefault, first);
+ assertSame(cache.canonicalize(changedDefault), changedDefault);
+ assertEquals(nestedValue(first).getDefaultNullValue(), -1);
+
+ Schema changedType = complexSchema();
+ nestedValue(changedType).setDataType(DataType.LONG);
+ assertEquals(changedType, changedDefault);
+ assertSame(cache.canonicalize(changedType), changedType);
+
+ Schema removedChild = complexSchema();
+ ((ComplexFieldSpec) removedChild.getFieldSpecFor("nested")).getChildFieldSpecs().remove("value");
+ assertSame(cache.canonicalize(removedChild), removedChild);
+ }
+
+ @Test
+ public void testConcurrentEqualSchemasShareOneInstance()
+ throws Exception {
+ TableSchemaCache cache = new TableSchemaCache();
+ ExecutorService executor = Executors.newFixedThreadPool(8);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ List<Future<Schema>> results = new ArrayList<>();
+ for (int i = 0; i < 32; i++) {
+ Schema fresh = schema();
+ results.add(executor.submit(() -> {
+ assertTrue(start.await(10, TimeUnit.SECONDS));
+ return cache.canonicalize(fresh);
+ }));
+ }
+ start.countDown();
+ Schema shared = results.get(0).get(10, TimeUnit.SECONDS);
+ for (Future<Schema> result : results) {
+ assertSame(result.get(10, TimeUnit.SECONDS), shared);
+ }
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testTimestampNormalizationPrecedesSchemaSharing() {
+ TableSchemaCache cache = new TableSchemaCache();
+ TableConfig firstTable = timestampTable(TimestampIndexGranularity.DAY);
+ IndexLoadingConfig first = new IndexLoadingConfig(null, firstTable, schema(), cache::canonicalize);
+ Schema shared = first.getSchema();
+ assertTrue(shared.hasColumn("$ts$DAY"));
+ FieldSpec derived = shared.getFieldSpecFor("$ts$DAY");
+
+ TableConfig freshTable = timestampTable(TimestampIndexGranularity.DAY);
+ IndexLoadingConfig second = new IndexLoadingConfig(null, freshTable, schema(), cache::canonicalize);
+ assertSame(second.getSchema(), shared);
+ assertSame(second.getSchema().getFieldSpecFor("$ts$DAY"), derived);
+ assertTrue(second.getFieldIndexConfigByColName().get("$ts$DAY").getConfig(StandardIndexes.range()).isEnabled());
+ assertEquals(freshTable.getIndexingConfig().getRangeIndexColumns(), List.of("$ts$DAY"));
+ assertEquals(freshTable.getIngestionConfig().getTransformConfigs().size(), 1);
+
+ IndexLoadingConfig changed = new IndexLoadingConfig(null, timestampTable(TimestampIndexGranularity.HOUR), schema(),
+ cache::canonicalize);
+ assertNotSame(changed.getSchema(), shared);
+ assertTrue(changed.getSchema().hasColumn("$ts$HOUR"));
+ assertFalse(changed.getSchema().hasColumn("$ts$DAY"));
+ assertFalse(shared.hasColumn("$ts$HOUR"));
+ assertSame(shared.getFieldSpecFor("$ts$DAY"), derived);
+ }
+
+ private static Schema schema() {
+ return new Schema.SchemaBuilder().setSchemaName("testTable")
+ .addSingleValueDimension("id", DataType.INT, -1)
+ .addDateTime("ts", DataType.TIMESTAMP, "TIMESTAMP", "1:MILLISECONDS").build();
+ }
+
+ private static Schema complexSchema() {
+ Schema schema = schema();
+ ComplexFieldSpec child = new ComplexFieldSpec("value", DataType.MAP, true,
+ Map.of("key", new DimensionFieldSpec("key", DataType.STRING, true),
+ "value", new DimensionFieldSpec("value", DataType.INT, true, -1)));
+ schema.addField(new ComplexFieldSpec("nested", DataType.MAP, true,
+ Map.of("key", new DimensionFieldSpec("key", DataType.STRING, true), "value", child)));
+ return schema;
+ }
+
+ private static FieldSpec nestedValue(Schema schema) {
+ return ((ComplexFieldSpec) ((ComplexFieldSpec) schema.getFieldSpecFor("nested")).getChildFieldSpec("value"))
+ .getChildFieldSpec("value");
+ }
+
+ private static TableConfig timestampTable(TimestampIndexGranularity granularity) {
+ return new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable")
+ .setFieldConfigList(List.of(new FieldConfig.Builder("ts")
+ .withTimestampConfig(new TimestampConfig(List.of(granularity))).build())).build();
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
index 724bd04ef69..5340c0c32d7 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
@@ -228,6 +228,8 @@ public class ImmutableSegmentLoader {
segmentMetadata.removeColumn(column);
}
}
+ // Preprocessing must compare the stored definitions before matching specs can be shared with the table.
+ segmentMetadata.reuseFieldSpecs(schema);
} else {
indexLoadingConfig.addKnownColumns(columnMetadataMap.keySet());
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java
index 3782f3abbf8..d27576e82b3 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java
@@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.UnaryOperator;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.segment.local.segment.index.loader.columnminmaxvalue.ColumnMinMaxValueGeneratorMode;
@@ -51,6 +52,8 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.TimestampIndexUtils;
+import static java.util.Objects.requireNonNull;
+
/// Table level index loading config.
public class IndexLoadingConfig {
@@ -98,9 +101,19 @@ public class IndexLoadingConfig {
/// TODO: Revisit the init handling. Currently it doesn't apply tiered config override
public IndexLoadingConfig(@Nullable InstanceDataManagerConfig instanceDataManagerConfig,
@Nullable TableConfig tableConfig, @Nullable Schema schema) {
+ this(instanceDataManagerConfig, tableConfig, schema, UnaryOperator.identity());
+ }
+
+ /// Normalizes the supplied schema before choosing a shared instance. The canonicalizer must return an equivalent
+ /// schema that callers treat as read-only. Timestamp-index expansion only modifies the supplied inputs.
+ public IndexLoadingConfig(@Nullable InstanceDataManagerConfig instanceDataManagerConfig,
+ @Nullable TableConfig tableConfig, @Nullable Schema schema, UnaryOperator<Schema> schemaCanonicalizer) {
_instanceDataManagerConfig = instanceDataManagerConfig;
_tableConfig = tableConfig;
- _schema = schema;
+ if (tableConfig != null && schema != null) {
+ TimestampIndexUtils.applyTimestampIndex(tableConfig, schema);
+ }
+ _schema = schema != null ? requireNonNull(schemaCanonicalizer.apply(schema)) : null;
init();
}
@@ -171,10 +184,6 @@ public class IndexLoadingConfig {
}
private void extractFromTableConfigAndSchema() {
- if (_schema != null) {
- TimestampIndexUtils.applyTimestampIndex(_tableConfig, _schema);
- }
-
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
String tableReadMode = indexingConfig.getLoadMode();
if (tableReadMode != null) {
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SchemaFieldSpecReuseTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SchemaFieldSpecReuseTest.java
new file mode 100644
index 00000000000..09e98e3a909
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SchemaFieldSpecReuseTest.java
@@ -0,0 +1,215 @@
+/**
+ * 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.index.loader;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.config.SchemaSerDeUtils;
+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.ImmutableSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+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.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+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.assertFalse;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Exercises table FieldSpec sharing through real segment preprocessing and query-serving index readers.
+public class SchemaFieldSpecReuseTest {
+ private File _tempDir;
+ private File _segmentDir;
+ private TableConfig _tableConfig;
+
+ @BeforeMethod
+ public void setUp()
+ throws Exception {
+ _tempDir = Files.createTempDirectory("SchemaFieldSpecReuseTest").toFile();
+ _tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("schemaReuse")
+ .setSegmentVersion("v3").setNullHandlingEnabled(true).build();
+ List<GenericRow> rows = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue("id", i);
+ row.putValue("tags", new String[]{"x", "y"});
+ row.putValue("score", i + 0.5);
+ row.putValue("ts", 1000L + i);
+ if (i == 0) {
+ row.putDefaultNullValue("id", -1);
+ }
+ rows.add(row);
+ }
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, newSchema());
+ config.setOutDir(_tempDir.getAbsolutePath());
+ config.setSegmentName("segment");
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+ _segmentDir = driver.getOutputDirectory();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void tearDown() {
+ FileUtils.deleteQuietly(_tempDir);
+ }
+
+ @Test
+ public void testSharedSchemaAcrossConcurrentLoads()
+ throws Exception {
+ Schema schema = newSchema();
+ List<ImmutableSegment> segments = new ArrayList<>();
+ try {
+ try (var executor = Executors.newFixedThreadPool(4)) {
+ List<Future<ImmutableSegment>> futures = new ArrayList<>();
+ for (int i = 0; i < 8; i++) {
+ futures.add(executor.submit(() -> ImmutableSegmentLoader.load(_segmentDir,
+ new IndexLoadingConfig(_tableConfig, schema), false)));
+ }
+ for (Future<ImmutableSegment> future : futures) {
+ segments.add(future.get());
+ }
+ }
+ for (ImmutableSegment segment : segments) {
+ assertNotSame(segment.getSegmentMetadata().getSchema(), schema);
+ for (FieldSpec spec : schema.getAllFieldSpecs()) {
+ assertSame(segment.getSegmentMetadata().getColumnMetadataFor(spec.getName()).getFieldSpec(), spec);
+ assertSame(segment.getSegmentMetadata().getSchema().getFieldSpecFor(spec.getName()), spec);
+ }
+ assertValues(segment);
+ }
+ assertNotSame(segments.get(0).getSegmentMetadata().getSchema(), segments.get(1).getSegmentMetadata().getSchema());
+ assertEquals(schema.size(), 4, "Virtual columns must stay in each segment's own schema");
+ } finally {
+ for (ImmutableSegment segment : segments) {
+ segment.destroy();
+ }
+ }
+ }
+
+ @Test
+ public void testPhysicalDefaultMismatchAfterPreprocessing()
+ throws Exception {
+ Schema schema = newSchema();
+ schema.getFieldSpecFor("id").setDefaultNullValue(-2);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(_segmentDir, new IndexLoadingConfig(_tableConfig, schema));
+ try {
+ FieldSpec stored = segment.getSegmentMetadata().getColumnMetadataFor("id").getFieldSpec();
+ assertFalse(segment.getSegmentMetadata().getColumnMetadataFor("id").isAutoGenerated());
+ assertEquals(stored.getDefaultNullValue(), -1);
+ assertNotSame(stored, schema.getFieldSpecFor("id"));
+ assertTrue(segment.getDataSource("id").getDictionary().indexOf((int) stored.getDefaultNullValue()) >= 0);
+ assertEquals(segment.getDataSource("id").getDictionary().indexOf(-2), -1);
+ assertValues(segment);
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ @Test
+ public void testAutogeneratedDefaultIsRebuiltBeforeReuse()
+ throws Exception {
+ Schema original = newSchema();
+ original.addField(new DimensionFieldSpec("added", DataType.INT, true, -10));
+ ImmutableSegment first = ImmutableSegmentLoader.load(_segmentDir, new IndexLoadingConfig(_tableConfig, original));
+ try {
+ assertTrue(first.getSegmentMetadata().getColumnMetadataFor("added").isAutoGenerated());
+ assertSame(first.getSegmentMetadata().getColumnMetadataFor("added").getFieldSpec(),
+ original.getFieldSpecFor("added"));
+ assertDefaultColumn(first, -10);
+ } finally {
+ first.destroy();
+ }
+ Schema updated = SchemaSerDeUtils.fromZNRecord(SchemaSerDeUtils.toZNRecord(original));
+ updated.getFieldSpecFor("added").setDefaultNullValue(-20);
+ ImmutableSegment second = ImmutableSegmentLoader.load(_segmentDir, new IndexLoadingConfig(_tableConfig, updated));
+ try {
+ assertSame(second.getSegmentMetadata().getColumnMetadataFor("added").getFieldSpec(),
+ updated.getFieldSpecFor("added"));
+ assertDefaultColumn(second, -20);
+ assertEquals(original.getFieldSpecFor("added").getDefaultNullValue(), -10);
+ assertValues(second);
+ } finally {
+ second.destroy();
+ }
+ }
+
+ @Test
+ public void testReadWithoutTableSchema()
+ throws Exception {
+ ImmutableSegment segment = ImmutableSegmentLoader.load(_segmentDir, ReadMode.mmap);
+ try {
+ assertValues(segment);
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ private static Schema newSchema() {
+ return new Schema.SchemaBuilder().setSchemaName("schemaReuse")
+ .addSingleValueDimension("id", DataType.INT, -1).addMultiValueDimension("tags", DataType.STRING)
+ .addMetric("score", DataType.DOUBLE)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ }
+
+ private static void assertDefaultColumn(ImmutableSegment segment, int expected)
+ throws Exception {
+ try (PinotSegmentColumnReader reader = new PinotSegmentColumnReader(segment, "added")) {
+ for (int i = 0; i < 4; i++) {
+ assertEquals(reader.getValue(i), expected);
+ }
+ }
+ }
+
+ private static void assertValues(ImmutableSegment segment)
+ throws Exception {
+ try (PinotSegmentColumnReader ids = new PinotSegmentColumnReader(segment, "id");
+ PinotSegmentColumnReader tags = new PinotSegmentColumnReader(segment, "tags");
+ PinotSegmentColumnReader scores = new PinotSegmentColumnReader(segment, "score");
+ PinotSegmentColumnReader times = new PinotSegmentColumnReader(segment, "ts")) {
+ for (int i = 0; i < 4; i++) {
+ assertEquals(ids.getValue(i), i == 0 ? -1 : i);
+ assertEquals((Object[]) tags.getValue(i), new String[]{"x", "y"});
+ assertEquals(scores.getValue(i), i + 0.5);
+ assertEquals(times.getValue(i), 1000L + i);
+ }
+ }
+ assertTrue(segment.getDataSource("id").getNullValueVector().getNullBitmap().contains(0));
+ }
+}
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 747c6d8b14f..24e6612864b 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
@@ -70,8 +70,9 @@ import static com.google.common.base.Preconditions.checkElementIndex;
/// 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. 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. Callers must treat shared specs and their nested
-/// values as read-only. Deserialize [FieldSpec#toJsonObject()] to make a copy before editing a spec.
+/// one instance per distinct spec instead of retaining its own. Final segment loading can replace an equal spec
+/// with the table schema's instance. Callers must treat shared specs and their nested values as read-only.
+/// Deserialize [FieldSpec#toJsonObject()] to make a copy before editing a spec.
@SuppressWarnings({"rawtypes", "unchecked"})
public class ColumnMetadataImpl implements ColumnMetadata {
private static final long SIZE_MASK = 0xffffffffffffL;
@@ -83,7 +84,7 @@ public class ColumnMetadataImpl implements ColumnMetadata {
/// any of them and is released once the last one is unloaded. Thread-safe.
private static final Interner<FieldSpec> FIELD_SPEC_INTERNER = Interners.newWeakInterner();
- private final FieldSpec _fieldSpec;
+ private FieldSpec _fieldSpec;
private final int _totalDocs;
private final int _cardinality;
private final boolean _hasDictionary;
@@ -153,6 +154,11 @@ public class ColumnMetadataImpl implements ColumnMetadata {
return _fieldSpec;
}
+ /// Replaces an equal spec during final loading, before the segment is published.
+ void setFieldSpec(FieldSpec fieldSpec) {
+ _fieldSpec = fieldSpec;
+ }
+
@Override
public int getTotalDocs() {
return _totalDocs;
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java
index 7a09ad5243b..28bb21187e3 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java
@@ -57,6 +57,8 @@ import org.apache.pinot.segment.spi.index.startree.StarTreeV2Metadata;
import org.apache.pinot.segment.spi.store.ColumnIndexUtils;
import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
+import org.apache.pinot.spi.data.ComplexFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.env.CommonsConfigurationUtils;
import org.apache.pinot.spi.utils.CommonConstants.Segment.BuiltInVirtualColumn;
@@ -384,6 +386,60 @@ public class SegmentMetadataImpl implements SegmentMetadata {
return _segmentVersion;
}
+ /// Reuses equal table-schema field specs after preprocessing, before constructing readers or publishing the segment.
+ /// Mismatched specs retain their segment-specific values. Complex specs are excluded because their equality does
+ /// not compare children. The segment schema remains independently owned and keeps its existing field order.
+ @SuppressWarnings("deprecation") // Preserve legacy TIME specs and the segment schema's field order.
+ public void reuseFieldSpecs(@Nullable Schema tableSchema) {
+ if (tableSchema == null || _totalDocs == 0 || _columnMetadataMap == null || _columnMetadataMap.isEmpty()) {
+ return;
+ }
+ Map<String, FieldSpec> replacements = new HashMap<>();
+ for (Map.Entry<String, ColumnMetadata> entry : _columnMetadataMap.entrySet()) {
+ ColumnMetadata metadata = entry.getValue();
+ if (!(metadata instanceof ColumnMetadataImpl)) {
+ continue;
+ }
+ FieldSpec segmentSpec = metadata.getFieldSpec();
+ if (segmentSpec instanceof ComplexFieldSpec || ((ColumnMetadataImpl) metadata).getParentColumn() != null
+ || !entry.getKey().equals(segmentSpec.getName())) {
+ continue;
+ }
+ FieldSpec tableSpec = tableSchema.getFieldSpecFor(segmentSpec.getName());
+ if (tableSpec != segmentSpec && segmentSpec.equals(tableSpec)) {
+ replacements.put(segmentSpec.getName(), tableSpec);
+ }
+ }
+ if (replacements.isEmpty()) {
+ return;
+ }
+
+ // The collection returned by getAllFieldSpecs() is sorted by name, unlike the per-type lists.
+ List<FieldSpec> orderedSpecs = new ArrayList<>(_schema.size());
+ orderedSpecs.addAll(_schema.getDimensionFieldSpecs());
+ orderedSpecs.addAll(_schema.getMetricFieldSpecs());
+ if (_schema.getTimeFieldSpec() != null) {
+ orderedSpecs.add(_schema.getTimeFieldSpec());
+ }
+ orderedSpecs.addAll(_schema.getDateTimeFieldSpecs());
+ orderedSpecs.addAll(_schema.getComplexFieldSpecs());
+ for (Map.Entry<String, FieldSpec> entry : replacements.entrySet()) {
+ ColumnMetadataImpl metadata = (ColumnMetadataImpl) _columnMetadataMap.remove(entry.getKey());
+ FieldSpec fieldSpec = entry.getValue();
+ metadata.setFieldSpec(fieldSpec);
+ _columnMetadataMap.put(fieldSpec.getName(), metadata);
+ if (entry.getKey().equals(_timeColumn)) {
+ _timeColumn = fieldSpec.getName();
+ }
+ }
+ for (FieldSpec fieldSpec : orderedSpecs) {
+ _schema.removeField(fieldSpec.getName());
+ }
+ for (FieldSpec fieldSpec : orderedSpecs) {
+ _schema.addField(replacements.getOrDefault(fieldSpec.getName(), fieldSpec));
+ }
+ }
+
@Override
public Schema getSchema() {
return _schema;
diff --git a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataFieldSpecReuseTest.java b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataFieldSpecReuseTest.java
new file mode 100644
index 00000000000..92d225e5fb9
--- /dev/null
+++ b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataFieldSpecReuseTest.java
@@ -0,0 +1,252 @@
+/**
+ * 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.spi.index.metadata;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+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.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.Schema;
+import org.apache.pinot.spi.data.TimeFieldSpec;
+import org.apache.pinot.spi.data.TimeGranularitySpec;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+
+
+/// Verifies final-load field-spec reuse without changing segment metadata, schema order, or fallback values.
+@SuppressWarnings("deprecation") // Verify reuse preserves legacy TIME columns.
+public class SegmentMetadataFieldSpecReuseTest {
+ @Test
+ public void reusesEqualSpecsWithoutChangingMetadataOrSchemaOrder()
+ throws Exception {
+ List<FieldSpec> fields = List.of(new DimensionFieldSpec(new String("z"), DataType.INT, true),
+ new DimensionFieldSpec(new String("a"), DataType.INT, true),
+ new MetricFieldSpec(new String("count"), DataType.LONG),
+ new DateTimeFieldSpec(new String("date"), DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS"),
+ new TimeFieldSpec(new TimeGranularitySpec(DataType.INT, TimeUnit.HOURS, new String("time"))));
+ SegmentMetadataImpl metadata = metadata(10, fields);
+ Schema tableSchema = schema(fields);
+ Schema segmentSchema = metadata.getSchema();
+ FieldSpec firstDimension = segmentSchema.getDimensionFieldSpecs().get(0);
+ segmentSchema.removeField(firstDimension.getName());
+ segmentSchema.addField(firstDimension);
+ segmentSchema.setSchemaName("segment-schema");
+ segmentSchema.setEnableColumnBasedNullHandling(true);
+ segmentSchema.setPrimaryKeyColumns(List.of("z"));
+ List<String> dimensionOrder = new ArrayList<>(segmentSchema.getDimensionNames());
+ String metadataJson = JsonUtils.objectToString(metadata);
+ String schemaJson = segmentSchema.toSingleLineJsonString();
+ String tableJson = tableSchema.toSingleLineJsonString();
+ Map<String, ColumnMetadata> originalColumns = Map.copyOf(metadata.getColumnMetadataMap());
+
+ metadata.reuseFieldSpecs(tableSchema);
+
+ assertSame(metadata.getSchema(), segmentSchema);
+ assertEquals(segmentSchema.getDimensionNames(), dimensionOrder);
+ assertEquals(segmentSchema.toSingleLineJsonString(), schemaJson);
+ assertEquals(JsonUtils.objectToString(metadata), metadataJson);
+ assertEquals(tableSchema.toSingleLineJsonString(), tableJson);
+ for (FieldSpec field : fields) {
+ ColumnMetadata column = metadata.getColumnMetadataFor(field.getName());
+ assertSame(column, originalColumns.get(field.getName()));
+ assertSame(column.getFieldSpec(), field);
+ assertSame(segmentSchema.getFieldSpecFor(field.getName()), field);
+ assertSame(metadata.getColumnMetadataMap().ceilingKey(field.getName()), field.getName());
+ assertSame(segmentSchema.getFieldSpecMap().ceilingKey(field.getName()), field.getName());
+ }
+ assertSame(metadata.getTimeColumn(), tableSchema.getTimeFieldSpec().getName());
+
+ metadata.reuseFieldSpecs(tableSchema);
+ assertEquals(segmentSchema.toSingleLineJsonString(), schemaJson);
+ assertEquals(JsonUtils.objectToString(metadata), metadataJson);
+ }
+
+ @Test
+ public void retainsMissingAndEvolvedSpecs()
+ throws Exception {
+ List<FieldSpec> fields = List.of(new DimensionFieldSpec("missing", DataType.INT, true),
+ new DimensionFieldSpec("default", DataType.INT, true, -1),
+ new DimensionFieldSpec("type", DataType.INT, true),
+ new DimensionFieldSpec("multiValue", DataType.INT, false),
+ new DimensionFieldSpec("notNull", DataType.INT, true),
+ new DimensionFieldSpec("fieldType", DataType.INT, true),
+ new TimeFieldSpec(new TimeGranularitySpec(DataType.INT, TimeUnit.HOURS, "time")));
+ SegmentMetadataImpl metadata = metadata(10, fields);
+ DimensionFieldSpec notNull = new DimensionFieldSpec("notNull", DataType.INT, true);
+ notNull.setNotNull(true);
+ Schema tableSchema = schema(List.of(new DimensionFieldSpec("default", DataType.INT, true, -2),
+ new DimensionFieldSpec("type", DataType.LONG, true),
+ new DimensionFieldSpec("multiValue", DataType.INT, true), notNull,
+ new MetricFieldSpec("fieldType", DataType.INT),
+ new TimeFieldSpec(new TimeGranularitySpec(DataType.INT, TimeUnit.DAYS, "time"))));
+ Map<String, FieldSpec> before = Map.copyOf(metadata.getSchema().getFieldSpecMap());
+ String json = JsonUtils.objectToString(metadata);
+
+ metadata.reuseFieldSpecs(tableSchema);
+
+ before.forEach((name, field) -> {
+ assertSame(metadata.getColumnMetadataFor(name).getFieldSpec(), field);
+ assertSame(metadata.getSchema().getFieldSpecFor(name), field);
+ });
+ assertEquals(JsonUtils.objectToString(metadata), json);
+ assertEquals(metadata.getColumnMetadataFor("default").getFieldSpec().getDefaultNullValue(), -1);
+ }
+
+ @Test
+ public void skipsComplexSpecsAndKeepsUnmatchedInternedFallbacks()
+ throws Exception {
+ ComplexFieldSpec complex = new ComplexFieldSpec("nested", DataType.OPEN_STRUCT, true,
+ Map.of("child", new DimensionFieldSpec("nested$$child", DataType.INT, true)));
+ List<FieldSpec> fields = List.of(complex, new DimensionFieldSpec("matched", DataType.INT, true),
+ new DimensionFieldSpec("missing", DataType.INT, true));
+ SegmentMetadataImpl first = metadata(10, fields);
+ SegmentMetadataImpl second = metadata(10, fields);
+ FieldSpec originalComplex = first.getColumnMetadataFor("nested").getFieldSpec();
+ ComplexFieldSpec evolved = new ComplexFieldSpec("nested", DataType.OPEN_STRUCT, true,
+ Map.of("other", new DimensionFieldSpec("nested$$other", DataType.LONG, true)));
+ assertEquals(originalComplex, evolved, "Complex equality does not compare children");
+ DimensionFieldSpec matched = new DimensionFieldSpec("matched", DataType.INT, true);
+ Schema tableSchema = schema(List.of(evolved, matched));
+ assertNotSame(first.getColumnMetadataFor("matched").getFieldSpec(), matched);
+
+ first.reuseFieldSpecs(tableSchema);
+ second.reuseFieldSpecs(tableSchema);
+
+ assertSame(first.getColumnMetadataFor("nested").getFieldSpec(), originalComplex);
+ assertSame(first.getSchema().getFieldSpecFor("nested"), originalComplex);
+ assertEquals(((ComplexFieldSpec) originalComplex).getChildFieldSpecs().keySet(),
+ complex.getChildFieldSpecs().keySet());
+ assertSame(first.getColumnMetadataFor("matched").getFieldSpec(), matched);
+ assertSame(second.getColumnMetadataFor("matched").getFieldSpec(), matched);
+ assertSame(first.getColumnMetadataFor("missing").getFieldSpec(),
+ second.getColumnMetadataFor("missing").getFieldSpec());
+ }
+
+ @Test
+ public void skipsEmptyAndConsumingMetadataAndAbsentSchema()
+ throws Exception {
+ Schema tableSchema = schema(List.of(new DimensionFieldSpec("empty", DataType.INT, true)));
+ SegmentMetadataImpl empty = metadata(0, new ArrayList<>(tableSchema.getAllFieldSpecs()));
+ FieldSpec original = empty.getColumnMetadataFor("empty").getFieldSpec();
+ empty.reuseFieldSpecs(tableSchema);
+ assertSame(empty.getColumnMetadataFor("empty").getFieldSpec(), original);
+
+ SegmentMetadataImpl consuming = new SegmentMetadataImpl("table", "consuming", tableSchema, 0);
+ consuming.reuseFieldSpecs(tableSchema);
+ assertSame(consuming.getSchema(), tableSchema);
+
+ SegmentMetadataImpl regular = metadata(10, new ArrayList<>(tableSchema.getAllFieldSpecs()));
+ FieldSpec regularSpec = regular.getColumnMetadataFor("empty").getFieldSpec();
+ regular.reuseFieldSpecs(null);
+ assertSame(regular.getColumnMetadataFor("empty").getFieldSpec(), regularSpec);
+ }
+
+ @Test
+ public void doesNotMatchMaterializedChildAgainstTopLevelColumn()
+ throws Exception {
+ SegmentMetadataImpl metadata = metadata(10, List.of(new DimensionFieldSpec("cpu", DataType.INT, true, -1)));
+ FieldSpec topLevel = metadata.getColumnMetadataFor("cpu").getFieldSpec();
+ DimensionFieldSpec child = new DimensionFieldSpec("cpu", DataType.INT, true, 0);
+ ColumnMetadataImpl childMetadata = ColumnMetadataImpl.builder().setFieldSpec(child).setTotalDocs(10).build();
+ metadata.getColumnMetadataMap().put("metrics$cpu", childMetadata);
+ Schema tableSchema = schema(List.of(new DimensionFieldSpec("cpu", DataType.INT, true, 0)));
+
+ metadata.reuseFieldSpecs(tableSchema);
+
+ assertSame(metadata.getColumnMetadataFor("cpu").getFieldSpec(), topLevel);
+ assertSame(metadata.getSchema().getFieldSpecFor("cpu"), topLevel);
+ assertSame(metadata.getColumnMetadataFor("metrics$cpu"), childMetadata);
+ assertSame(childMetadata.getFieldSpec(), child);
+ assertEquals(topLevel.getDefaultNullValue(), -1);
+ }
+
+ private static Schema schema(List<FieldSpec> fields) {
+ Schema schema = new Schema();
+ fields.forEach(schema::addField);
+ return schema;
+ }
+
+ private static SegmentMetadataImpl metadata(int totalDocs, List<FieldSpec> fields)
+ throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty(Segment.SEGMENT_NAME, "reuse-test");
+ properties.setProperty(Segment.SEGMENT_TOTAL_DOCS, Integer.toString(totalDocs));
+ Map<FieldType, String> lists = Map.of(FieldType.DIMENSION, Segment.DIMENSIONS, FieldType.METRIC, Segment.METRICS,
+ FieldType.TIME, Segment.TIME_COLUMN_NAME, FieldType.DATE_TIME, Segment.DATETIME_COLUMNS,
+ FieldType.COMPLEX, Segment.COMPLEX_COLUMNS);
+ lists.forEach((type, key) -> {
+ String names = fields.stream().filter(field -> field.getFieldType() == type).map(FieldSpec::getName)
+ .collect(Collectors.joining(","));
+ if (!names.isEmpty()) {
+ properties.setProperty(key, names);
+ }
+ });
+ fields.forEach(field -> writeField(properties, field.getName(), field));
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ properties.store(stream, null);
+ byte[] creationMetadata = ByteBuffer.allocate(16).putLong(1).putLong(2).array();
+ return new SegmentMetadataImpl(new ByteArrayInputStream(stream.toByteArray()),
+ new ByteArrayInputStream(creationMetadata));
+ }
+
+ private static void writeField(Properties properties, String column, FieldSpec field) {
+ properties.setProperty(Column.getKeyFor(column, Column.COLUMN_NAME), field.getName());
+ properties.setProperty(Column.getKeyFor(column, Column.COLUMN_TYPE), field.getFieldType().name());
+ properties.setProperty(Column.getKeyFor(column, Column.DATA_TYPE), field.getDataType().name());
+ properties.setProperty(Column.getKeyFor(column, Column.IS_SINGLE_VALUED),
+ Boolean.toString(field.isSingleValueField()));
+ properties.setProperty(Column.getKeyFor(column, Column.CARDINALITY), "1");
+ if (field instanceof ComplexFieldSpec) {
+ Map<String, FieldSpec> children = ((ComplexFieldSpec) field).getChildFieldSpecs();
+ properties.setProperty(Column.getKeyFor(column, Column.COMPLEX_CHILD_FIELD_NAMES),
+ String.join(",", children.keySet()));
+ children.forEach((name, child) -> writeField(properties, ComplexFieldSpec.getFullChildName(column, name), child));
+ } else {
+ properties.setProperty(Column.getKeyFor(column, Column.DEFAULT_NULL_VALUE), field.getDefaultNullValueString());
+ }
+ if (field instanceof DateTimeFieldSpec) {
+ properties.setProperty(Column.getKeyFor(column, Column.DATETIME_FORMAT), ((DateTimeFieldSpec) field).getFormat());
+ properties.setProperty(Column.getKeyFor(column, Column.DATETIME_GRANULARITY),
+ ((DateTimeFieldSpec) field).getGranularity());
+ }
+ if (field instanceof TimeFieldSpec) {
+ properties.setProperty(Segment.TIME_UNIT,
+ ((TimeFieldSpec) field).getIncomingGranularitySpec().getTimeType().name());
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org