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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
