diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 3e02f3b6d238..ab704dc61cbb 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -864,6 +864,7 @@ You can configure automatic compaction through the following properties: |Property|Description|Required| |--------|-----------|--------| +|`type`|Where the compaction schema comes from. `inline` declares it in this configuration, using the schema properties below. `catalog` reads it from the datasource's [catalog](../development/extensions-core/catalog.md) table definition on every run, and accepts none of the schema properties; see [Catalog-based compaction](../development/extensions-core/catalog.md#catalog-based-compaction).|no (default = `inline`)| |`dataSource`|The datasource name to be compacted.|yes| |`taskPriority`|[Priority](../ingestion/tasks.md#lock-priority) of compaction task.|no (default = 25)| |`inputSegmentSizeBytes`|Maximum number of total segment bytes processed per compaction task. Since a time chunk must be processed in its entirety, if the segments for a particular time chunk have a total size in bytes greater than this parameter, compaction will not run for that time chunk.|no (default = 100,000,000,000,000 i.e. 100TB)| diff --git a/docs/data-management/automatic-compaction.md b/docs/data-management/automatic-compaction.md index 0d84a81bcdcb..7afefb23e57b 100644 --- a/docs/data-management/automatic-compaction.md +++ b/docs/data-management/automatic-compaction.md @@ -40,7 +40,13 @@ This topic guides you through setting up automatic compaction for your Druid clu ## Auto-compaction syntax You can configure automatic compaction dynamically without restarting Druid. -The automatic compaction system uses the following syntax: + +A configuration's `type` selects where the schema it compacts toward comes from. The default, `inline`, declares the +schema in the configuration itself; `catalog` reads it from the datasource's catalog table definition. + +### Define the schema inline + +An `inline` schema declaration for the automatic compaction uses the following syntax: ```json { @@ -58,7 +64,7 @@ The automatic compaction system uses the following syntax: } ``` -Most fields in the auto-compaction configuration correlate to a typical [Druid ingestion spec](../ingestion/ingestion-spec.md). +Most fields in the configuration correlate to a typical [Druid ingestion spec](../ingestion/ingestion-spec.md). The following properties only apply to auto-compaction: * `skipOffsetFromLatest` * `skipIntervals` @@ -68,7 +74,7 @@ The following properties only apply to auto-compaction: Since the automatic compaction system provides a management layer on top of manual compaction tasks, the auto-compaction configuration does not include task-specific properties found in a typical Druid ingestion spec. The following properties are automatically set by the auto-compaction system: -* `type`: Set to `compact`. +* `type`: Set to `compact`. This is the generated task's type, not the configuration's own `type` described above. * `id`: Generated using the task type, datasource name, interval, and timestamp. The task ID is prefixed with `auto`. * `context`: Set according to the user-provided `taskContext`. @@ -82,6 +88,28 @@ maximize performance and minimize disk usage of the `compact` tasks launched by For more details on each of the specs in an auto-compaction configuration, see [Automatic compaction dynamic configuration](../configuration/index.md#automatic-compaction-dynamic-configuration). +### Take the schema from the catalog + +An [inline](#define-the-schema-inline) configuration declares its own schema, which has to be kept in step with the +datasource by hand. If the datasource has a [catalog](../development/extensions-core/catalog.md) table definition, +compaction can read the schema from it instead, by setting `"type": "catalog"`: + +```json +{ + "type": "catalog", + "dataSource": "events", + "engine": "msq" +} +``` + +Such a configuration carries no schema of its own. It accepts only `dataSource`, `engine`, `skipOffsetFromLatest`, +`skipIntervals`, `taskPriority`, `taskContext`, and `inputSegmentSizeBytes`; everything else (segment granularity, +partitioning, projections, and the stored column layout) is read from the table definition on every compaction run. +Editing the definition therefore changes how the datasource is compacted from then on, which means it can also rewrite +data that is already stored. See +[Catalog-based compaction](../development/extensions-core/catalog.md#catalog-based-compaction) for exactly which table +properties take effect and which are ignored. + ## Auto-compaction using compaction supervisors :::info diff --git a/docs/development/extensions-core/catalog.md b/docs/development/extensions-core/catalog.md index 20842c927ae4..f1a6ceb3d28d 100644 --- a/docs/development/extensions-core/catalog.md +++ b/docs/development/extensions-core/catalog.md @@ -46,7 +46,7 @@ data is written into a particular column of the table. ### Effects of a table definition A table definition takes effect when it is read, not when it is written: changing one, whether through SQL DDL or the -REST API, rewrites nothing. What it affects: +REST API, rewrites nothing by itself. What it affects: - **SELECT queries** report each declared column with its declared type, in declared order. Segments that store a column as a different physical type are converted to the declared type at query time. Columns present in segments @@ -60,6 +60,9 @@ REST API, rewrites nothing. What it affects: override, such as with its own `PARTITIONED BY`. - **Streaming and native batch ingestion** do not consult the catalog; their specs are unaffected by any table definition. +- **Existing segments** are rewritten to match the definition only if the datasource is configured for + [catalog-based compaction](#catalog-based-compaction), which is off unless you set it up. Without it, a definition + never changes data that is already stored. ### SQL DDL @@ -67,9 +70,11 @@ Tables can be defined with SQL instead of by posting a table specification. `CRE submitted to the Broker like any other SQL statement, and write the same catalog metadata the REST API does. They return no rows. -These statements change catalog metadata only. They never create, modify, or delete segments: defining a table does -not ingest anything, and altering a column does not rewrite existing data. Changes take effect for subsequent queries -and ingestion, as described in [Effects of a table definition](#effects-of-a-table-definition). +These statements change catalog metadata only. They do not themselves create, modify, or delete segments: defining a +table does not ingest anything, and altering a column does not rewrite existing data. Changes take effect for +subsequent queries and ingestion, as described in [Effects of a table definition](#effects-of-a-table-definition). If +the datasource is configured for [catalog-based compaction](#catalog-based-compaction), compaction later rewrites +existing segments to match the new definition in the background. These statements are disabled by default. Set `druid.sql.planner.enableCatalogDdl` to `true` on the Broker to enable them. They require both `READ` and `WRITE` permission on the datasource, the same permissions the catalog API @@ -228,7 +233,8 @@ Unlike an aggregate projection, a `__base` body cannot filter or group: the base It is the only projection that chooses a clustering. `ALTER TABLE ... ADD PROJECTION __base AS ( ... )` gives an existing table a layout, and -`ALTER TABLE ... DROP PROJECTION __base` removes it. Both affect future segments only. +`ALTER TABLE ... DROP PROJECTION __base` removes it. Both affect future segments, plus any existing segments that +[catalog-based compaction](#catalog-based-compaction) later rewrites. Every other name beginning with `__` remains reserved. @@ -249,6 +255,66 @@ There is no `DROP TABLE`. Deleting a table's catalog entry without deleting its for the statement, so removing a specification is left to the [delete API](#delete-a-table) until the semantics are settled. +### Catalog-based compaction + +By default, a table definition only shapes data on its way in, and segments written before the definition changed keep +whatever they were written with. However, [Automatic compaction](../../data-management/automatic-compaction.md) can be +configured to use the catalog, so that it rewrites existing segments toward the current definition as it compacts them. +A datasource opts in with a compaction config of type `catalog`, which reads the live table definition on every +compaction run rather than repeating the schema in the compaction config: + +```json +{ + "type": "catalog", + "dataSource": "events", + "engine": "msq" +} +``` + +Once this is configured, editing the table definition becomes a background data change: the next compaction run +notices that existing segments no longer match the definition and rewrites them. The edit is still not immediate, and +it only reaches intervals that compaction actually visits. + +Not every [table property](#table-properties) drives compaction. These do: + +| Property | Effect on existing segments | +|---|---| +| `segmentGranularity` | Segments are re-partitioned in time to the declared granularity. | +| `clusterKeys` | Segments are range-partitioned by the key columns, in the declared order. All key columns must be `VARCHAR`; compaction fails on a numeric one. | +| `targetSegmentRows` | Rewritten segments are sized to this row count, the same size an `INSERT` produces. | +| `projections` | Once the table declares any [projection](#projections), the declared set is authoritative: those are built and any others are dropped. A table that declares none leaves whatever the segments already have. | +| `baseTable` | The table's physical layout, written in SQL as the [`__base` projection](#the-base-table). The stored column order, clustering, and computed columns are rewritten to match it, along with the declared column list it combines with. | +| `sealed` | On a `__base` table, decides whether columns the table does not declare are kept. See below. | + +Everything else a table can declare is ignored by compaction today, including `hiddenColumns` and, on a table with no +`__base` projection, the column list itself. A plain catalog table's columns therefore never trim: compaction infers +its schema from the segments it is rewriting, so a column dropped from the definition survives in the data and stays +queryable. + +Dropping a column trims it from storage only on a table that has both a `__base` projection and `SEALED`: + +| Table | Effect of dropping a declared column | +|---|---| +| No `__base` projection | Column is kept. Compaction takes its schema from the segments. | +| `__base`, not sealed | Column is kept. Compaction reads the segments for columns the definition omits and appends them after the declared ones, the same as at ingest. | +| `__base` with `SEALED` | **Column is trimmed.** The declaration is treated as complete, so the rewritten segments no longer store it. | + +That last row is the case to be careful with: on a sealed clustered table, `ALTER TABLE ... DROP COLUMN` eventually +deletes data, just not at the moment the statement runs. + +The other column statements behave the same way — nothing on a plain table, a background rewrite on a `__base` table: + +- `ADD COLUMN` changes nothing by itself either way. The column has no data in existing segments, and compaction does + not invent any. +- `ALTER COLUMN ... SET DATA TYPE` rewrites the column to the declared type on a `__base` table, since the declared + types are what the layout is built from. On a plain table the stored type is left alone and the declared type + continues to apply at query time only. +- Reordering a `__base` body, or changing which columns it clusters by, re-sorts and re-groups the rows of every + segment compaction rewrites. + +Catalog-based compaction requires the [MSQ task engine](../../multi-stage-query/index.md) for a table with a `__base` +projection; the native engine rejects the layout. + ### API Objects #### TableSpec @@ -266,7 +332,12 @@ A tableSpec defines a table | PropertyKeyName | PropertyValueType | Description | Required | Default | |----------------------|-------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|---------| | `segmentGranularity` | String | determines how time-based partitioning is done. See [Partitioning by time](../../multi-stage-query/concepts.md#partitioning-by-time). Can specify any of the values as permitted for [PARTITIONED BY](../../multi-stage-query/reference.md#partitioned-by). This property value may be overridden at query time, by specifying the PARTITIONED BY clause. | no | null | -| `sealed` | boolean | require all columns in the table schema to be fully declared before data is ingested. Setting this to true will cause failure when DML queries attempt to add undefined columns to the table. | no | false | +| `targetSegmentRows` | Integer | the number of rows to aim for in each segment. Applied unless the query sets `msqRowsPerSegment` itself. If unset, the system default is used. | no | null | +| `clusterKeys` | List<Object\> | the columns each new segment is clustered by, as objects of the form `{"column": "", "desc": false}`. Acts as a default that an individual statement may override with its own CLUSTERED BY clause. Only ascending order is supported; a key with `"desc": true` is rejected. | no | null | +| `hiddenColumns` | List<String\> | names of columns present in the segments to hide from the SQL layer, so that unwanted columns can be removed from the table's schema without rewriting existing segments. The data remains stored. `__time` cannot be hidden. | no | null | +| `sealed` | boolean | require all columns in the table schema to be fully declared before data is ingested. Setting this to true will cause failure when DML queries attempt to add undefined columns to the table. On a table with a `__base` projection it also decides whether [compaction](#catalog-based-compaction) keeps undeclared columns. | no | false | +| `projections` | List<Object\> | the table's [projections](#projections), as objects of the form `{"spec": {}}`. Usually written with `ADD PROJECTION` rather than set directly. | no | null | +| `baseTable` | Object | the table's physical layout, described in SQL as the [`__base` projection](#the-base-table). Combines with the declared column list, which remains the source of truth for column names, types, and order. Usually written with `ADD PROJECTION __base` rather than set directly. | no | null | #### ColumnSpec diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CatalogBaseTableCompactionTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CatalogBaseTableCompactionTest.java new file mode 100644 index 000000000000..199d1022f40e --- /dev/null +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CatalogBaseTableCompactionTest.java @@ -0,0 +1,344 @@ +/* + * 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.druid.testing.embedded.compact; + +import org.apache.druid.catalog.guice.CatalogClientModule; +import org.apache.druid.catalog.guice.CatalogCoordinatorModule; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.catalog.model.TableMetadata; +import org.apache.druid.catalog.model.table.ClusterKeySpec; +import org.apache.druid.catalog.model.table.DatasourceDefn; +import org.apache.druid.catalog.model.table.TableBuilder; +import org.apache.druid.common.utils.IdUtils; +import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec; +import org.apache.druid.data.input.impl.DimensionSchema; +import org.apache.druid.data.input.impl.LongDimensionSchema; +import org.apache.druid.data.input.impl.StringDimensionSchema; +import org.apache.druid.data.input.impl.TimestampSpec; +import org.apache.druid.indexer.CompactionEngine; +import org.apache.druid.indexer.granularity.SegmentGranularitySpec; +import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec; +import org.apache.druid.indexing.common.task.TaskBuilder; +import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexSupervisorTask; +import org.apache.druid.indexing.compact.CompactionSupervisorSpec; +import org.apache.druid.indexing.overlord.Segments; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.DruidMetrics; +import org.apache.druid.rpc.UpdateResponse; +import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.server.coordinator.CatalogDataSourceCompactionConfig; +import org.apache.druid.server.coordinator.ClusterCompactionConfig; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedHistorical; +import org.apache.druid.testing.embedded.EmbeddedIndexer; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.catalog.TestCatalogClient; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.apache.druid.timeline.CompactionState; +import org.apache.druid.timeline.DataSegment; +import org.hamcrest.Matchers; +import org.joda.time.Period; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +/** + * End-to-end coverage of catalog-driven compaction of a clustered base table: the catalog's {@code baseTable} layout + * drives the compacted segment's physical shape, and a table that is not {@code sealed} makes compaction analyze the + * segments it is rewriting so columns the catalog does not declare survive. + */ +public class CatalogBaseTableCompactionTest extends EmbeddedClusterTestBase +{ + /** + * Declared by the catalog table, so the base table spec owns it. + */ + private static final String DECLARED_COLUMN = "channel"; + /** + * Present in the segments but absent from the catalog table, so only an analyze step can find it. + */ + private static final String UNDECLARED_COLUMN = "countryName"; + + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.catalog.client.maxSyncRetries", "0") + .addProperty("druid.manager.segments.pollDuration", "PT1s") + .addProperty("druid.manager.segments.useIncrementalCache", "always"); + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.manager.segments.useIncrementalCache", "always"); + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.catalog.client.pollingPeriod", "100"); + // MSQ sizes its memory budget per worker slot, so the indexer needs headroom to run the compaction controller. + private final EmbeddedIndexer indexer = new EmbeddedIndexer() + .setServerMemory(2_000_000_000L) + .addProperty("druid.worker.capacity", "4"); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster + .withEmbeddedDerbyAndZookeeper() + .useLatchableEmitter() + .useDefaultTimeoutForLatchableEmitter(120) + .addExtensions(CatalogClientModule.class, CatalogCoordinatorModule.class) + // Clustered base-table segments require the V10 segment format. + .addCommonProperty("druid.indexer.task.buildV10", "true") + .addServer(coordinator) + .addServer(overlord) + .addServer(indexer) + .addServer(broker) + .addServer(new EmbeddedHistorical()); + } + + @Test + public void test_compactClusteredBaseTable_withUnsealedCatalogTable_preservesUndeclaredColumns() throws IOException + { + ingestClusteredSegment(); + + // The catalog declares only two of the three stored columns and says nothing about being sealed, so it accepts + // columns it does not declare -- exactly the case where compaction has to go looking. + createCatalogTable(false); + configureCompaction(); + runCompaction(); + + final CompactionState compactionState = Assertions.assertDoesNotThrow( + () -> getOnlyUsedSegment().getLastCompactionState() + ); + Assertions.assertNotNull(compactionState); + Assertions.assertNotNull(compactionState.getBaseTable()); + + final List columns = compactionState.getBaseTable() + .getDimensionsSpec() + .getDimensions() + .stream() + .map(DimensionSchema::getName) + .collect(Collectors.toList()); + // The declared shape leads, and the column the catalog knows nothing about was carried over rather than dropped. + Assertions.assertEquals( + List.of(DECLARED_COLUMN, ColumnHolder.TIME_COLUMN_NAME, UNDECLARED_COLUMN), + columns + ); + Assertions.assertEquals( + List.of(DECLARED_COLUMN), + ((ClusteredValueGroupsBaseTableProjectionSpec) compactionState.getBaseTable()).getClusteringColumnNames() + ); + } + + @Test + public void test_compactClusteredBaseTable_withSealedCatalogTable_dropsUndeclaredColumns() throws IOException + { + ingestClusteredSegment(); + + // A sealed table declares everything, so the spec is taken as authoritative and the undeclared column is not + // carried over. This is the counterpart of the test above: the same segments, the same layout, one property apart. + createCatalogTable(true); + configureCompaction(); + runCompaction(); + + final CompactionState compactionState = getOnlyUsedSegment().getLastCompactionState(); + Assertions.assertNotNull(compactionState); + Assertions.assertEquals( + List.of(DECLARED_COLUMN, ColumnHolder.TIME_COLUMN_NAME), + compactionState.getBaseTable() + .getDimensionsSpec() + .getDimensions() + .stream() + .map(DimensionSchema::getName) + .collect(Collectors.toList()) + ); + } + + @Test + public void test_compactClusteredBaseTable_withClusterKeys_partitionsByRange() throws IOException + { + ingestClusteredSegment(); + + // Cluster keys and a base table layout are different axes: the keys range-partition rows across segments, the + // base table groups them into cluster groups within a segment. A table that declares both gets both. + createCatalogTable(false, DECLARED_COLUMN); + configureCompaction(); + runCompaction(); + + final CompactionState compactionState = getOnlyUsedSegment().getLastCompactionState(); + Assertions.assertNotNull(compactionState); + Assertions.assertInstanceOf(DimensionRangePartitionsSpec.class, compactionState.getPartitionsSpec()); + Assertions.assertEquals( + List.of(DECLARED_COLUMN), + ((DimensionRangePartitionsSpec) compactionState.getPartitionsSpec()).getPartitionDimensions() + ); + // targetSegmentRows is used verbatim, not scaled by 1.5x, matching how an INSERT into this table sizes segments. + Assertions.assertEquals(1000, compactionState.getPartitionsSpec().getMaxRowsPerSegment()); + // The base table layout still applies, undeclared column and all. + Assertions.assertEquals( + List.of(DECLARED_COLUMN, ColumnHolder.TIME_COLUMN_NAME, UNDECLARED_COLUMN), + compactionState.getBaseTable() + .getDimensionsSpec() + .getDimensions() + .stream() + .map(DimensionSchema::getName) + .collect(Collectors.toList()) + ); + } + + /** + * Posts the catalog compaction supervisor and waits until the coordinator reports nothing left awaiting compaction + * for this datasource. + *

+ * That wait is also the convergence assertion: the coordinator recomputes {@code interval/waitCompact/count} from + * the recorded {@link CompactionState} against the config on every run, so if the appended columns made the + * compacted segment look permanently out of date, the metric would never reach zero and this would time out rather + * than pass. + */ + private void runCompaction() + { + cluster.callApi().postSupervisor( + new CompactionSupervisorSpec( + new CatalogDataSourceCompactionConfig( + dataSource, + // baseTable requires the MSQ compaction engine + CompactionEngine.MSQ, + Period.ZERO, + null, + null, + null, + null, + null + ), + false, + null + ) + ); + + overlord.latchableEmitter().waitForEvent( + event -> event.hasMetricName("task/run/time") + .hasDimension(DruidMetrics.TASK_TYPE, "compact") + .hasDimension(DruidMetrics.DATASOURCE, dataSource) + ); + overlord.latchableEmitter().waitForEvent( + event -> event.hasMetricName("interval/waitCompact/count") + .hasDimension(DruidMetrics.DATASOURCE, dataSource) + .hasValueMatching(Matchers.equalTo(0L)) + ); + } + + private void createCatalogTable(boolean sealed, String... clusterKeyColumns) + { + final TableBuilder builder = + TableBuilder.datasource(dataSource, "HOUR") + .column(DECLARED_COLUMN, Columns.SQL_VARCHAR) + .column(Columns.TIME_COLUMN, Columns.LONG) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(List.of(DECLARED_COLUMN), null, null)); + if (sealed) { + builder.property(DatasourceDefn.SEALED_PROPERTY, true); + } + if (clusterKeyColumns.length > 0) { + // Size the segments too, so the recorded range spec is the sized one: if the config and what MSQ records + // disagreed on how targetSegmentRows maps onto a range spec, the interval would never read as compacted and + // runCompaction() below would spin rather than settle. + builder.property(DatasourceDefn.TARGET_SEGMENT_ROWS_PROPERTY, 1000); + builder.clusterColumns( + Arrays.stream(clusterKeyColumns) + .map(column -> new ClusterKeySpec(column, false)) + .toArray(ClusterKeySpec[]::new) + ); + } + final TableMetadata table = builder.build(); + new TestCatalogClient(cluster).createTable(table, true); + } + + private void configureCompaction() + { + final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord( + o -> o.updateClusterCompactionConfig( + new ClusterCompactionConfig(1.0, 10, null, true, CompactionEngine.MSQ, true) + ) + ); + Assertions.assertTrue(updateResponse.isSuccess()); + } + + /** + * Ingests a clustered V10 segment carrying {@link #UNDECLARED_COLUMN} in addition to what the catalog table below + * declares, so compaction has a column to discover. + */ + private void ingestClusteredSegment() throws IOException + { + final File inputFile = new File(cluster.getTestFolder().newFolder(), "clustered-input.json"); + Files.write( + inputFile.toPath(), + ("{\"time\":\"2024-01-01T00:10:00Z\",\"channel\":\"#en\",\"countryName\":\"US\"}\n" + + "{\"time\":\"2024-01-01T00:20:00Z\",\"channel\":\"#en\",\"countryName\":\"CA\"}\n" + + "{\"time\":\"2024-01-01T00:30:00Z\",\"channel\":\"#fr\",\"countryName\":\"FR\"}\n") + .getBytes(StandardCharsets.UTF_8) + ); + + final String taskId = IdUtils.getRandomId(); + final ParallelIndexSupervisorTask task = TaskBuilder + .ofTypeIndexParallel() + .jsonInputFormat() + .localInputSourceWithFiles(inputFile) + .dataSchema( + builder -> builder + .withDataSource(dataSource) + .withTimestamp(new TimestampSpec("time", "iso", null)) + .withSegmentGranularity( + new SegmentGranularitySpec(Granularities.HOUR, List.of(Intervals.of("2024-01-01/2024-01-02"))) + ) + .withBaseTable( + ClusteredValueGroupsBaseTableProjectionSpec + .builder() + .columns( + new StringDimensionSchema(DECLARED_COLUMN), + new LongDimensionSchema(ColumnHolder.TIME_COLUMN_NAME), + new StringDimensionSchema(UNDECLARED_COLUMN) + ) + .clusteringColumns(DECLARED_COLUMN) + .build() + ) + ) + .tuningConfig(t -> t.withMaxNumConcurrentSubTasks(1)) + .withId(taskId); + + cluster.callApi().onLeaderOverlord(o -> o.runTask(taskId, task)); + cluster.callApi().waitForTaskToSucceed(taskId, overlord); + cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); + } + + private DataSegment getOnlyUsedSegment() + { + final List segments = overlord + .bindings() + .segmentsMetadataStorage() + .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE) + .stream() + .filter(segment -> !segment.isTombstone()) + .collect(Collectors.toList()); + Assertions.assertEquals(1, segments.size(), segments::toString); + return segments.get(0); + } +} diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java index 0e857e527862..23cab04f872b 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java @@ -101,6 +101,7 @@ import org.apache.druid.segment.loading.AcquireSegmentAction; import org.apache.druid.segment.loading.SegmentCacheManager; import org.apache.druid.segment.projections.AggregateProjectionSchema; +import org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema; import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager; import org.apache.druid.segment.transform.CompactionTransformSpec; import org.apache.druid.segment.transform.TransformSpec; @@ -175,6 +176,7 @@ public class CompactionTask extends AbstractBatchIndexTask implements PendingSeg private final List projections; @Nullable private final BaseTableProjectionSpec baseTable; + private final boolean sealed; @Nullable private final CompactionTuningConfig tuningConfig; @Nullable @@ -200,6 +202,7 @@ public CompactionTask( @JsonProperty("granularitySpec") @Nullable final ClientCompactionTaskGranularitySpec granularitySpec, @JsonProperty("baseTable") @Nullable final BaseTableProjectionSpec baseTable, @JsonProperty("projections") @Nullable List projections, + @JsonProperty("sealed") @Nullable final Boolean sealed, @JsonProperty("tuningConfig") @Nullable final TuningConfig tuningConfig, @JsonProperty("context") @Nullable final Map context, @JsonProperty("compactionRunner") final CompactionRunner compactionRunner, @@ -272,6 +275,7 @@ public CompactionTask( } this.projections = projections; this.baseTable = baseTable; + this.sealed = sealed == null || sealed; this.tuningConfig = tuningConfig != null ? getTuningConfig(tuningConfig) : null; this.segmentProvider = new SegmentProvider(dataSource, this.ioConfig.getInputSpec()); // Note: The default compactionRunnerType used here should match the default runner used in CompactSegments#run @@ -445,6 +449,16 @@ public BaseTableProjectionSpec getBaseTable() return baseTable; } + /** + * Whether {@link #getBaseTable()} is the complete declaration of the datasource schema. When false, the existing + * segments are analyzed and any column the spec does not declare is preserved rather than dropped. + */ + @JsonProperty + public boolean isSealed() + { + return sealed; + } + @JsonProperty public CompactionRunner getCompactionRunner() { @@ -568,6 +582,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception granularitySpec, projections, baseTable, + sealed, getMetricBuilder(), this.identifyMultiValuedDimensions() ); @@ -604,6 +619,7 @@ static Map createInputDataSchemas( @Nullable final ClientCompactionTaskGranularitySpec granularitySpec, @Nullable final List projections, @Nullable final BaseTableProjectionSpec baseTable, + final boolean sealed, final ServiceMetricEvent.Builder metricBuilder, boolean needMultiValuedColumns ) throws IOException @@ -700,6 +716,7 @@ static Map createInputDataSchemas( : granularitySpec.withSegmentGranularity(segmentGranularityToUse), projections, baseTable, + sealed, needMultiValuedColumns ); final QuerySegmentSpec querySegmentSpec; @@ -749,6 +766,7 @@ static Map createInputDataSchemas( granularitySpec, projections, baseTable, + sealed, needMultiValuedColumns ); final QuerySegmentSpec querySegmentSpec; @@ -798,6 +816,7 @@ private static DataSchema createDataSchema( @Nonnull ClientCompactionTaskGranularitySpec granularitySpec, @Nullable List projections, @Nullable BaseTableProjectionSpec baseTable, + boolean sealed, boolean needMultiValuedColumns ) { @@ -805,9 +824,28 @@ private static DataSchema createDataSchema( // BaseTable mode: the spec owns dimensions/metrics and (via its query-granularity virtual column) query // granularity; segment granularity + intervals are the compaction config's. Attach the config's query // granularity up front (a no-op if absent/NONE); the generator's incremental index applies it at build time. - // Existing-segment analysis (dims/metrics/rollup inference) is bypassed — the spec is authoritative. - final BaseTableProjectionSpec baseTableSpec = - baseTable.withQueryGranularity(granularitySpec.getQueryGranularity()); + // A sealed spec is the complete declaration of the table, so existing-segment analysis has nothing to add and is + // skipped entirely. + BaseTableProjectionSpec baseTableSpec = baseTable.withQueryGranularity(granularitySpec.getQueryGranularity()); + if (!sealed) { + // Otherwise the segments may carry columns the spec does not know about. Find them and append them after the + // declared columns, leaving the shape the operator asked for unchanged. Only dimensions are carried over; a + // spec that stores metrics does not exist yet, and asking the analyzer for them would fail compaction on + // segments whose aggregators cannot be merged even though nothing would store them. + final ExistingSegmentAnalyzer analyzer = new ExistingSegmentAnalyzer( + segments, + false, + false, + true, + false, + false, + false + ); + fetchAndProcess(analyzer, emitter, metricBuilder); + baseTableSpec = baseTableSpec.withAdditionalColumns( + undeclaredColumns(baseTableSpec, analyzer.getDimensionsSpec()) + ); + } return CombinedDataSchema.forBaseTable( dataSource, new TimestampSpec(ColumnHolder.TIME_COLUMN_NAME, "millis", null), @@ -832,20 +870,7 @@ private static DataSchema createDataSchema( needMultiValuedColumns ); - final Stopwatch stopwatch = Stopwatch.createStarted(); - try { - existingSegmentAnalyzer.fetchAndProcessIfNeeded(); - } - finally { - if (emitter != null) { - emitter.emit( - metricBuilder.setMetric( - "compact/segmentAnalyzer/fetchAndProcessMillis", - stopwatch.millisElapsed() - ) - ); - } - } + fetchAndProcess(existingSegmentAnalyzer, emitter, metricBuilder); final Granularity queryGranularityToUse; if (granularitySpec.getQueryGranularity() == null) { @@ -899,6 +924,52 @@ private static DataSchema createDataSchema( ); } + /** + * The dimensions of {@code analyzed} that {@code baseTable} does not already declare, in the order the analyzer + * found them. Columns the spec cannot store (the time column, anything a virtual column computes) are left for + * {@link BaseTableProjectionSpec#withAdditionalColumns} to reject rather than dropped silently here. + */ + private static List undeclaredColumns( + BaseTableProjectionSpec baseTable, + DimensionsSpec analyzed + ) + { + final Set declared = new HashSet<>(); + for (DimensionSchema declaredColumn : baseTable.getDimensionsSpec().getDimensions()) { + declared.add(declaredColumn.getName()); + } + return analyzed.getDimensions() + .stream() + .filter(dimension -> !declared.contains(dimension.getName())) + .collect(Collectors.toList()); + } + + /** + * Runs the analyzer, timing the segment download and processing so operators can see what schema inference is + * costing them. The analyzer itself short-circuits when nothing it was asked for requires reading the segments. + */ + private static void fetchAndProcess( + ExistingSegmentAnalyzer analyzer, + @Nullable ServiceEmitter emitter, + ServiceMetricEvent.Builder metricBuilder + ) + { + final Stopwatch stopwatch = Stopwatch.createStarted(); + try { + analyzer.fetchAndProcessIfNeeded(); + } + finally { + if (emitter != null) { + emitter.emit( + metricBuilder.setMetric( + "compact/segmentAnalyzer/fetchAndProcessMillis", + stopwatch.millisElapsed() + ) + ); + } + } + } + /** * Lazily fetch and load {@link QueryableIndex}, skipping tombstones. */ @@ -1042,15 +1113,32 @@ public void fetchAndProcessIfNeeded() processRollup(index); processQueryGranularity(index); - processDimensionsSpec(index); + // Column data is read from a different index for a clustered segment, but rollup, query granularity, + // metrics and projections all come from the top-level metadata either way. + final QueryableIndex columnIndex = columnIndex(index); + processDimensionsSpec(columnIndex); processMetricsSpec(index); - processMultiValuedDimensions(index); + processMultiValuedDimensions(columnIndex); processProjections(index); } } } } + /** + * The index to read column data from. A clustered segment stores its rows in per-cluster-group containers and its + * top-level index deliberately holds no columns of its own, so reading the schema from it would find nothing; + * every group carries the same schema, so the first one answers for the segment. + */ + private static QueryableIndex columnIndex(QueryableIndex index) + { + final ClusteredValueGroupsBaseTableSchema clusteredBaseTable = index.getMetadata().getClusteredBaseTable(); + if (clusteredBaseTable == null || clusteredBaseTable.getClusterGroups().isEmpty()) { + return index; + } + return index.getClusterGroupQueryableIndex(clusteredBaseTable.getClusterGroups().get(0), true); + } + public Boolean getRollup() { if (!needRollup) { @@ -1437,6 +1525,7 @@ public static class Builder private List projections; @Nullable private BaseTableProjectionSpec baseTable; + private Boolean sealed; public Builder( String dataSource, @@ -1546,6 +1635,12 @@ public Builder baseTable(@Nullable BaseTableProjectionSpec baseTable) return this; } + public Builder sealed(@Nullable Boolean sealed) + { + this.sealed = sealed; + return this; + } + public CompactionTask build() { return new CompactionTask( @@ -1563,6 +1658,7 @@ public CompactionTask build() granularitySpec, baseTable, projections, + sealed, tuningConfig, context, compactionRunner, diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java index ec5e6d9b5183..e73b557eaa79 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java @@ -73,6 +73,8 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.HashMap; import java.util.Map; @@ -183,6 +185,7 @@ public void testClientCompactionTaskQueryBaseTableSerde() throws IOException null, baseTable, null, + null, ImmutableMap.of(), new ClientCompactionRunnerInfo(CompactionEngine.MSQ) ); @@ -192,6 +195,47 @@ public void testClientCompactionTaskQueryBaseTableSerde() throws IOException Assertions.assertEquals(baseTable, actual.getBaseTable()); Assertions.assertEquals(query, actual); + // A query that says nothing about sealing declares a complete schema, which is what every pre-existing payload + // means. + Assertions.assertTrue(actual.isSealed()); + } + + @Test + public void testClientCompactionTaskQuerySealedSerde() throws IOException + { + // 'sealed' has to survive the round trip onto the task's own field, and a payload that omits it (every + // pre-existing one) has to mean a complete declaration. + Assertions.assertFalse(roundTripSealed(false)); + Assertions.assertTrue(roundTripSealed(true)); + Assertions.assertTrue(roundTripSealed(null)); + } + + private boolean roundTripSealed(@Nullable Boolean sealed) throws IOException + { + final ClientCompactionTaskQuery query = new ClientCompactionTaskQuery( + "id", + "datasource", + new ClientCompactionIOConfig( + new ClientCompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds"), + true + ), + null, + null, + null, + null, + null, + null, + null, + sealed, + ImmutableMap.of(), + new ClientCompactionRunnerInfo(CompactionEngine.NATIVE) + ); + + final byte[] json = MAPPER.writeValueAsBytes(query); + final boolean fromQuery = ((ClientCompactionTaskQuery) MAPPER.readValue(json, ClientTaskQuery.class)).isSealed(); + final boolean fromTask = ((CompactionTask) MAPPER.readValue(json, Task.class)).isSealed(); + Assertions.assertEquals(fromQuery, fromTask, "query and task must agree on 'sealed'"); + return fromTask; } private static ObjectMapper setupInjectablesInObjectMapper(ObjectMapper objectMapper) @@ -370,6 +414,7 @@ private ClientCompactionTaskQuery createCompactionTaskQuery(String id, Compactio transformSpec, null, null, + null, context, new ClientCompactionRunnerInfo(CompactionEngine.NATIVE) ); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java index aa4e1c5a6e14..fd594cbc7c89 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java @@ -29,6 +29,7 @@ import com.fasterxml.jackson.databind.module.SimpleModule; import com.google.common.base.Preconditions; import com.google.common.base.Supplier; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; import com.google.common.collect.Lists; import com.google.common.collect.Maps; @@ -40,6 +41,7 @@ import org.apache.druid.common.guava.SettableSupplier; import org.apache.druid.data.input.InputSource; import org.apache.druid.data.input.impl.AggregateProjectionSpec; +import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec; import org.apache.druid.data.input.impl.DimensionSchema; import org.apache.druid.data.input.impl.DimensionsSpec; import org.apache.druid.data.input.impl.DoubleDimensionSchema; @@ -339,6 +341,19 @@ private static List findDimensions(int startIndex, Interval segmentInter return dimensions; } + /** + * A clustered base table over two of the columns the test segments carry, so a non-sealed compaction has real + * undeclared columns to discover. + */ + private static final ClusteredValueGroupsBaseTableProjectionSpec TENANT_BASE_TABLE = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .columns( + new StringDimensionSchema("string_dim_0"), + new LongDimensionSchema(ColumnHolder.TIME_COLUMN_NAME) + ) + .clusteringColumns("string_dim_0") + .build(); + private static CompactionTask.CompactionTuningConfig createTuningConfig() { return TuningConfigBuilder.forCompactionTask() @@ -792,6 +807,7 @@ public void testCreateIngestionSchema() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -825,6 +841,77 @@ public void testCreateIngestionSchema() throws IOException ); } + @Test + public void testCreateIngestionSchemaWithSealedBaseTableSkipsSegmentAnalysis() throws IOException + { + // A sealed base table is the complete declaration of the datasource, so the schema is exactly what was declared + // and the existing segments are never read. + final DataSchema dataSchema = Iterables.getOnlyElement( + CompactionTask.createInputDataSchemas( + toolbox, + LockGranularity.TIME_CHUNK, + new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)), + null, + null, + null, + new ClientCompactionTaskGranularitySpec(Granularities.MONTH, null, null), + null, + TENANT_BASE_TABLE, + true, + METRIC_BUILDER, + false + ).values() + ); + + Assertions.assertEquals( + ImmutableList.of("string_dim_0", ColumnHolder.TIME_COLUMN_NAME), + dataSchema.getDimensionsSpec().getDimensionNames() + ); + Assertions.assertEquals(TENANT_BASE_TABLE, dataSchema.getBaseTable()); + } + + @Test + public void testCreateIngestionSchemaWithUnsealedBaseTableAppendsUndeclaredColumns() throws IOException + { + // The declaration is partial, so the columns the segments carry but the spec does not declare are discovered and + // appended after the declared ones rather than dropped on the floor. + final DataSchema dataSchema = Iterables.getOnlyElement( + CompactionTask.createInputDataSchemas( + toolbox, + LockGranularity.TIME_CHUNK, + new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)), + null, + null, + null, + new ClientCompactionTaskGranularitySpec(Granularities.MONTH, null, null), + null, + TENANT_BASE_TABLE, + false, + METRIC_BUILDER, + false + ).values() + ); + + final List dimensions = dataSchema.getDimensionsSpec().getDimensionNames(); + // The declared shape is untouched: the clustering prefix still leads and __time keeps its declared position. + Assertions.assertEquals( + ImmutableList.of("string_dim_0", ColumnHolder.TIME_COLUMN_NAME), + dimensions.subList(0, 2) + ); + Assertions.assertEquals( + ImmutableList.of("string_dim_0"), + ((ClusteredValueGroupsBaseTableProjectionSpec) dataSchema.getBaseTable()).getClusteringColumnNames() + ); + // Every other column of the existing segments came along, each exactly once. + for (String dimension : findDimensions(0, SEGMENT_INTERVALS.get(0))) { + Assertions.assertTrue(dimensions.contains(dimension), dimension); + } + Assertions.assertEquals(new HashSet<>(dimensions).size(), dimensions.size(), dimensions.toString()); + // Only dimensions are carried over; the source segments' aggregators are not, since a clustered base table has no + // metric columns to put them in. + Assertions.assertEquals(0, dataSchema.getAggregators().length); + } + @Test public void testCreateIngestionSchemaWithTargetPartitionSize() throws IOException { @@ -856,6 +943,7 @@ public void testCreateIngestionSchemaWithTargetPartitionSize() throws IOExceptio null, null, null, + true, METRIC_BUILDER, false ); @@ -921,6 +1009,7 @@ public void testCreateIngestionSchemaWithMaxTotalRows() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -987,6 +1076,7 @@ public void testCreateIngestionSchemaWithNumShards() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -1060,6 +1150,7 @@ public void testCreateIngestionSchemaWithCustomDimensionsSpec() throws IOExcepti null, null, null, + true, METRIC_BUILDER, false ); @@ -1113,6 +1204,7 @@ public void testCreateIngestionSchemaWithCustomMetricsSpec() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -1159,6 +1251,7 @@ public void testCreateIngestionSchemaWithCustomSegments() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -1212,6 +1305,7 @@ public void testCreateIngestionSchemaWithDifferentSegmentSet() throws IOExceptio null, null, null, + true, METRIC_BUILDER, false ); @@ -1247,6 +1341,7 @@ public void testMissingMetadata() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -1292,6 +1387,7 @@ public void testSegmentGranularityAndNullQueryGranularity() throws IOException new ClientCompactionTaskGranularitySpec(new PeriodGranularity(Period.months(3), null, null), null, null), null, null, + true, METRIC_BUILDER, false ); @@ -1339,6 +1435,7 @@ public void testQueryGranularityAndNullSegmentGranularity() throws IOException new ClientCompactionTaskGranularitySpec(null, new PeriodGranularity(Period.months(3), null, null), null), null, null, + true, METRIC_BUILDER, false ); @@ -1387,6 +1484,7 @@ public void testQueryGranularityAndSegmentGranularityNonNull() throws IOExceptio ), null, null, + true, METRIC_BUILDER, false ); @@ -1437,6 +1535,7 @@ public void testNullGranularitySpec() throws IOException null, null, null, + true, METRIC_BUILDER, false ); @@ -1485,6 +1584,7 @@ public void testGranularitySpecWithNullQueryGranularityAndNullSegmentGranularity new ClientCompactionTaskGranularitySpec(null, null, null), null, null, + true, METRIC_BUILDER, false ); @@ -1533,6 +1633,7 @@ public void testGranularitySpecWithNotNullRollup() new ClientCompactionTaskGranularitySpec(null, null, true), null, null, + true, METRIC_BUILDER, false ); @@ -1566,6 +1667,7 @@ public void testGranularitySpecWithNullRollup() new ClientCompactionTaskGranularitySpec(null, null, null), null, null, + true, METRIC_BUILDER, false ); @@ -1601,6 +1703,7 @@ public void testMultiValuedDimensionsProcessing() new ClientCompactionTaskGranularitySpec(null, null, null), null, null, + true, METRIC_BUILDER, true ); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java index c3bcdcab0b02..1a4b21ecb2d5 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java @@ -450,6 +450,7 @@ public CompactionTask withId(String taskId) granularitySpec, null, null, + null, tuningConfig.build(), null, null, diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java b/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java index 292a26fe68cc..1c7adb793586 100644 --- a/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java +++ b/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java @@ -104,4 +104,19 @@ default boolean hasEqualCompactionState(BaseTableProjectionSpec other) { return equals(other); } + + /** + * Variant of {@link #hasEqualCompactionState(BaseTableProjectionSpec)} for a base table that is not the complete + * declaration of its table. When {@code allowUndeclaredColumns} is set, {@code other} is also equivalent if it is + * this spec extended by {@link #withAdditionalColumns}: those extra columns come from the segments rather than from + * the config, so comparing them strictly would make every segment look permanently out of date and recompact it on + * every run. + *

+ * The default ignores the relaxation and compares strictly; an implementation that can absorb undeclared columns + * overrides this to express what "extended by" means for its own state. + */ + default boolean hasEqualCompactionState(BaseTableProjectionSpec other, boolean allowUndeclaredColumns) + { + return hasEqualCompactionState(other); + } } diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java index 4a2855ba9b38..19563304c477 100644 --- a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java +++ b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java @@ -229,6 +229,30 @@ public boolean hasEqualCompactionState(BaseTableProjectionSpec other) .equals(((ClusteredValueGroupsBaseTableProjectionSpec) other).withoutQueryGranularity()); } + /** + * Since {@link #withAdditionalColumns} appends undeclared columns after the declared ones and changes nothing + * else, {@code other} is this spec extended exactly when it declares the same clustering columns and virtual + * columns and its column list starts with this one. List equality is the zero-appended-columns case, so this + * subsumes {@link #hasEqualCompactionState(BaseTableProjectionSpec)}. + */ + @Override + public boolean hasEqualCompactionState(BaseTableProjectionSpec other, boolean allowUndeclaredColumns) + { + if (!allowUndeclaredColumns) { + return hasEqualCompactionState(other); + } + if (!(other instanceof ClusteredValueGroupsBaseTableProjectionSpec)) { + return false; + } + final ClusteredValueGroupsBaseTableProjectionSpec mine = withoutQueryGranularity(); + final ClusteredValueGroupsBaseTableProjectionSpec theirs = + ((ClusteredValueGroupsBaseTableProjectionSpec) other).withoutQueryGranularity(); + return mine.clusteringColumns.equals(theirs.clusteringColumns) + && mine.virtualColumns.equals(theirs.virtualColumns) + && theirs.columns.size() >= mine.columns.size() + && theirs.columns.subList(0, mine.columns.size()).equals(mine.columns); + } + /** * Appends the given columns after the declared ones, which keeps the clustering columns the leading prefix of * {@link #getColumns()}. diff --git a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java index a09fdfc42de2..fd2d7a8dc43c 100644 --- a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java +++ b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java @@ -332,6 +332,59 @@ void testWithAdditionalColumnsAppendsAfterDeclaredColumns() Assertions.assertEquals(spec.getColumns(), spec.getDimensionsSpec().getDimensions()); } + @Test + void testHasEqualCompactionStateAllowingUndeclaredColumnsAcceptsAppendedColumns() + { + final ClusteredValueGroupsBaseTableProjectionSpec declared = tenantSpec(); + final ClusteredValueGroupsBaseTableProjectionSpec recorded = + declared.withAdditionalColumns(ImmutableList.of(new StringDimensionSchema("city"))); + + // What a non-sealed compaction records is a superset of what the config declares; comparing it strictly would + // recompact the datasource on every run. + Assertions.assertFalse(declared.hasEqualCompactionState(recorded)); + Assertions.assertFalse(declared.hasEqualCompactionState(recorded, false)); + Assertions.assertTrue(declared.hasEqualCompactionState(recorded, true)); + // Equality is the zero-appended-columns case. + Assertions.assertTrue(declared.hasEqualCompactionState(declared, true)); + // The relaxation is one-directional: dropping a column the config declares is still a change. + Assertions.assertFalse(recorded.hasEqualCompactionState(declared, true)); + } + + @Test + void testHasEqualCompactionStateAllowingUndeclaredColumnsStillComparesEverythingElse() + { + final ClusteredValueGroupsBaseTableProjectionSpec declared = tenantSpec(); + + // Query granularity is compared by its own check, so it is excluded from both sides here too. + Assertions.assertTrue( + declared.hasEqualCompactionState(declared.withQueryGranularity(Granularities.HOUR), true) + ); + // A changed declared column is a change even though the column count only grew. + final ClusteredValueGroupsBaseTableProjectionSpec reordered = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .columns( + new StringDimensionSchema("tenant"), + new LongDimensionSchema("__time"), + new StringDimensionSchema("region"), + new StringDimensionSchema("city") + ) + .clusteringColumns("tenant") + .build(); + Assertions.assertFalse(declared.hasEqualCompactionState(reordered, true)); + // A changed clustering column is a change. + final ClusteredValueGroupsBaseTableProjectionSpec reclustered = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("__time"), + new StringDimensionSchema("city") + ) + .clusteringColumns("tenant", "region") + .build(); + Assertions.assertFalse(declared.hasEqualCompactionState(reclustered, true)); + } + @Test void testWithAdditionalColumnsNullAndEmptyAreNoOps() { diff --git a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQuery.java b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQuery.java index da2052794271..63970e5ec099 100644 --- a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQuery.java +++ b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQuery.java @@ -55,6 +55,7 @@ public class ClientCompactionTaskQuery implements ClientTaskQuery private final List projections; @Nullable private final BaseTableProjectionSpec baseTable; + private final boolean sealed; private final Map context; private final ClientCompactionRunnerInfo compactionRunner; @@ -70,6 +71,7 @@ public ClientCompactionTaskQuery( @JsonProperty("transformSpec") CompactionTransformSpec transformSpec, @JsonProperty("baseTable") @Nullable BaseTableProjectionSpec baseTable, @JsonProperty("projections") @Nullable List projections, + @JsonProperty("sealed") @Nullable Boolean sealed, @JsonProperty("context") Map context, @JsonProperty("compactionRunner") @Nullable ClientCompactionRunnerInfo compactionRunner ) @@ -84,6 +86,7 @@ public ClientCompactionTaskQuery( this.transformSpec = transformSpec; this.projections = projections; this.baseTable = baseTable; + this.sealed = sealed == null || sealed; this.context = context; this.compactionRunner = compactionRunner; } @@ -168,6 +171,16 @@ public BaseTableProjectionSpec getBaseTable() return baseTable; } + /** + * Whether {@link #getBaseTable()} is the complete declaration of the datasource schema. False means the task must + * analyze the existing segments and preserve columns the spec does not declare. + */ + @JsonProperty("sealed") + public boolean isSealed() + { + return sealed; + } + @JsonProperty("compactionRunner") @Nullable public ClientCompactionRunnerInfo getCompactionRunner() @@ -203,6 +216,7 @@ public boolean equals(Object o) Objects.equals(transformSpec, that.transformSpec) && Objects.equals(projections, that.projections) && Objects.equals(baseTable, that.baseTable) && + sealed == that.sealed && Objects.equals(context, that.context) && Objects.equals(compactionRunner, that.compactionRunner); } @@ -220,6 +234,7 @@ public int hashCode() transformSpec, projections, baseTable, + sealed, context, compactionRunner ); @@ -239,8 +254,9 @@ public String toString() ", dimensionsSpec=" + dimensionsSpec + ", metricsSpec=" + Arrays.toString(metricsSpec) + ", transformSpec=" + transformSpec + - ", catalogConfig=" + projections + + ", projections=" + projections + ", baseTable=" + baseTable + + ", sealed=" + sealed + ", context=" + context + ", compactionRunner=" + compactionRunner + '}'; diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java index d5cadf98ed7a..bb283c6d606d 100644 --- a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java +++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java @@ -649,9 +649,16 @@ private CompactionStatus baseTableIsUpToDate(CompactionState lastCompactionState { // The baseTable spec compares its own state; segment/query granularity and rollup are covered by their own // checks, so the spec excludes them. A null configured baseTable means "don't care". + // + // When the config is not sealed, the compaction task appends the columns it finds in the segments to the spec, + // so the recorded state is a superset of the configured one and a strict comparison would recompact forever. + // The relaxed comparison accepts that superset. The cost is that the config-derived fingerprint never matches + // either, so these datasources always fall through to this field-by-field pass instead of the fingerprint fast + // path; recording the declared spec instead is not an option, since it would misdescribe the segment. final BaseTableProjectionSpec configured = compactionConfig.getBaseTable(); final BaseTableProjectionSpec current = lastCompactionState.getBaseTable(); - if (configured == null || (current != null && configured.hasEqualCompactionState(current))) { + if (configured == null + || (current != null && configured.hasEqualCompactionState(current, !compactionConfig.isSealed()))) { return COMPLETE; } return configChanged("baseTable", configured, current, String::valueOf); diff --git a/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java index a080ae01acd2..884da7be07d1 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java @@ -26,18 +26,24 @@ import com.google.common.base.Preconditions; import org.apache.druid.catalog.MetadataCatalog; import org.apache.druid.catalog.model.CatalogUtils; +import org.apache.druid.catalog.model.DatasourceBaseTableMetadata; import org.apache.druid.catalog.model.DatasourceProjectionMetadata; import org.apache.druid.catalog.model.ResolvedTable; import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.table.ClusterKeySpec; import org.apache.druid.catalog.model.table.DatasourceDefn; import org.apache.druid.client.indexing.ClientCompactionRunnerInfo; import org.apache.druid.common.config.Configs; import org.apache.druid.data.input.impl.AggregateProjectionSpec; import org.apache.druid.data.input.impl.BaseTableProjectionSpec; import org.apache.druid.indexer.CompactionEngine; +import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; +import org.apache.druid.indexer.partitions.PartitionsSpec; import org.apache.druid.java.util.common.granularity.Granularity; import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.segment.transform.CompactionTransformSpec; +import org.apache.druid.utils.CollectionUtils; import org.joda.time.Interval; import org.joda.time.Period; @@ -145,12 +151,49 @@ public Integer getMaxRowsPerSegment() return null; } + /** + * Partitioning comes from the catalog: {@link DatasourceDefn#CLUSTER_KEYS_PROPERTY} becomes the range partition + * dimensions and {@link DatasourceDefn#TARGET_SEGMENT_ROWS_PROPERTY} sizes the segments. A table that declares no + * cluster keys falls back to dynamic partitioning. + *

+ * A {@link #getBaseTable()} layout does not replace this: its clustering columns group rows within a + * segment, while cluster keys range-partition across segments, so a table that declares both gets both. + *

+ * This is not optional the way the other schema fields are: the MSQ engine rejects a compaction task with no + * partitionsSpec, and a base table layout can only be compacted by MSQ. + */ @JsonIgnore @Nullable @Override public UserCompactionTaskQueryTuningConfig getTuningConfig() { - return null; + final ResolvedTable table = catalog.resolveTable(tableId); + if (table == null) { + return null; + } + final Integer targetSegmentRows = table.decodeProperty(DatasourceDefn.TARGET_SEGMENT_ROWS_PROPERTY); + final List clusterKeys = getClusterKeys(table); + + final PartitionsSpec partitionsSpec; + if (CollectionUtils.isNullOrEmpty(clusterKeys)) { + partitionsSpec = new DynamicPartitionsSpec(targetSegmentRows, null); + } else { + // Range partitioning is ascending only, DatasourceDefn.ClusterKeysDefn already rejects a descending cluster key, + // so a catalog table cannot declare one. + partitionsSpec = new DimensionRangePartitionsSpec( + null, + targetSegmentRows == null ? PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT : targetSegmentRows, + clusterKeys.stream().map(ClusterKeySpec::expr).collect(Collectors.toList()), + false + ); + } + return UserCompactionTaskQueryTuningConfig.builder().partitionsSpec(partitionsSpec).build(); + } + + @Nullable + private List getClusterKeys(@Nullable ResolvedTable table) + { + return table == null ? null : table.decodeProperty(DatasourceDefn.CLUSTER_KEYS_PROPERTY); } @JsonIgnore @@ -170,9 +213,11 @@ public Granularity getSegmentGranularity() if (table == null) { return null; } - return CatalogUtils.asDruidGranularity( - table.decodeProperty(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY) - ); + final String segmentGranularity = table.decodeProperty(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY); + if (segmentGranularity == null) { + return null; + } + return CatalogUtils.asDruidGranularity(segmentGranularity); } @JsonIgnore @@ -203,12 +248,42 @@ public List getProjections() return projections.stream().map(DatasourceProjectionMetadata::getSpec).collect(Collectors.toList()); } + /** + * Derives the physical base table spec from the catalog table's {@link DatasourceDefn#BASE_TABLE_PROPERTY} layout, + * combined with its declared columns via {@link DatasourceBaseTableMetadata#createSpec}, the same way + * {@code MSQTaskQueryMaker} does for INSERT and REPLACE so that compaction lays segments out exactly as ingestion + * does. A table that declares no layout has no spec, and compaction falls back to the legacy schema fields. + */ @JsonIgnore @Nullable @Override public BaseTableProjectionSpec getBaseTable() { - return null; + final ResolvedTable table = catalog.resolveTable(tableId); + if (table == null) { + return null; + } + final DatasourceBaseTableMetadata baseTable = table.decodeProperty(DatasourceDefn.BASE_TABLE_PROPERTY); + if (baseTable == null) { + return null; + } + return baseTable.createSpec(table.spec().columns()); + } + + /** + * A catalog table is sealed only when it says so: an absent property means the table accepts columns it does not + * declare, which ingestion appends to the base table spec, so compaction must analyze the existing segments to find + * and preserve them rather than treating the declaration as complete. + */ + @JsonIgnore + @Override + public boolean isSealed() + { + final ResolvedTable table = catalog.resolveTable(tableId); + if (table == null) { + return true; + } + return Boolean.TRUE.equals(table.decodeProperty(DatasourceDefn.SEALED_PROPERTY)); } @JsonIgnore diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java index 1cad30913daf..4d6a6ea39aa9 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java @@ -116,6 +116,20 @@ public interface DataSourceCompactionConfig @Nullable AggregatorFactory[] getMetricsSpec(); + /** + * Whether the schema declared by this config is the complete schema of the datasource, i.e. the segments cannot + * contain columns it does not declare. When false, compaction analyzes the existing segments even where this config + * does declare a schema, and preserves whatever the declaration does not cover. + *

+ * Only {@link #getBaseTable()} observes this. The legacy {@link #getDimensionsSpec()} and {@link #getMetricsSpec()} + * express a partial declaration by being absent, so analysis for those is already driven by nullness; making a + * declaration that is present merge with an analyzed schema rather than replace it is not implemented. + */ + default boolean isSealed() + { + return true; + } + /** * Converts this compaction config to a {@link CompactionState}. *

diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java index a15de9933646..3197ce7136ab 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java @@ -403,6 +403,7 @@ public static ClientCompactionTaskQuery createCompactionTask( config.getMetricsSpec(), config.getTransformSpec(), config.getBaseTable(), + config.isSealed(), config.getProjections(), dropExisting, autoCompactionContext, @@ -460,6 +461,7 @@ private static ClientCompactionTaskQuery compactSegments( @Nullable AggregatorFactory[] metricsSpec, @Nullable CompactionTransformSpec transformSpec, @Nullable BaseTableProjectionSpec baseTable, + boolean sealed, @Nullable List projectionSpecs, @Nullable Boolean dropExisting, Map context, @@ -514,6 +516,7 @@ private static ClientCompactionTaskQuery compactSegments( transformSpec, baseTable, projectionSpecs, + sealed, context, compactionRunner ); diff --git a/server/src/test/java/org/apache/druid/server/compaction/CompactionSlotManagerTest.java b/server/src/test/java/org/apache/druid/server/compaction/CompactionSlotManagerTest.java index 7615fd19bb38..03d0247fe122 100644 --- a/server/src/test/java/org/apache/druid/server/compaction/CompactionSlotManagerTest.java +++ b/server/src/test/java/org/apache/druid/server/compaction/CompactionSlotManagerTest.java @@ -76,6 +76,7 @@ private ClientCompactionTaskQuery createMsqTask(Map context) null, null, null, + null, context, new ClientCompactionRunnerInfo(CompactionEngine.MSQ) ); diff --git a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java index cd625d191aaf..1b3ceb22fb0b 100644 --- a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java +++ b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java @@ -19,6 +19,12 @@ package org.apache.druid.server.compaction; +import com.google.common.collect.ImmutableList; +import org.apache.druid.catalog.MapMetadataCatalog; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.table.TableBuilder; import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig; import org.apache.druid.data.input.impl.AggregateProjectionSpec; import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec; @@ -44,6 +50,7 @@ import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.TestDataSource; import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnHolder; import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.data.CompressionStrategy; import org.apache.druid.segment.metadata.DefaultIndexingStateFingerprintMapper; @@ -53,6 +60,7 @@ import org.apache.druid.segment.nested.NestedCommonFormatColumnFormatSpec; import org.apache.druid.segment.transform.CompactionTransformSpec; import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.server.coordinator.CatalogDataSourceCompactionConfig; import org.apache.druid.server.coordinator.DataSourceCompactionConfig; import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig; import org.apache.druid.server.coordinator.UserCompactionTaskDimensionsConfig; @@ -79,6 +87,34 @@ public class CompactionStatusTest .size(100_000_000L) .build(); + private static final String SEALED_WIKI = TestDataSource.WIKI + "_sealed"; + private static final String UNSEALED_WIKI = TestDataSource.WIKI + "_unsealed"; + private static final MapMetadataCatalog CATALOG = new MapMetadataCatalog(new DefaultObjectMapper()); + + static { + // Same clustered layout under both names; the only difference is whether the table declares itself sealed. + final List columns = ImmutableList.of( + new ColumnSpec("tenant", ColumnType.STRING.asTypeString(), null), + new ColumnSpec("region", ColumnType.STRING.asTypeString(), null), + new ColumnSpec(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG.asTypeString(), null) + ); + CATALOG.addSpec( + TableId.datasource(SEALED_WIKI), + TableBuilder.datasource(SEALED_WIKI, "PT1H") + .columns(columns) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("tenant"), null, null)) + .sealed(true) + .buildSpec() + ); + CATALOG.addSpec( + TableId.datasource(UNSEALED_WIKI), + TableBuilder.datasource(UNSEALED_WIKI, "PT1H") + .columns(columns) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("tenant"), null, null)) + .buildSpec() + ); + } + private HeapMemoryIndexingStateStorage indexingStateStorage; private IndexingStateCache indexingStateCache; private IndexingStateFingerprintMapper fingerprintMapper; @@ -451,6 +487,104 @@ public void testStatusWhenBaseTableMatchesWithQueryGranularity() Assertions.assertTrue(status.isComplete()); } + @Test + public void testStatusWhenBaseTableHasUndeclaredColumnsAndCatalogTableIsNotSealed() + { + // A non-sealed table accepts columns it does not declare, so compaction appends what it finds in the segments and + // records a baseTable that is a superset of the declared one. Comparing that strictly would leave the datasource + // permanently "pending" and recompact it on every coordinator run. + final DataSourceCompactionConfig unsealed = catalogCompactionConfig(UNSEALED_WIKI); + final DataSegment segment = DataSegment + .builder(WIKI_SEGMENT) + .lastCompactionState(stateWithUndeclaredColumn(unsealed)) + .build(); + + Assertions.assertTrue( + CompactionStatus.compute(List.of(segment), unsealed, fingerprintMapper).isComplete() + ); + } + + @Test + public void testStatusWhenBaseTableHasUndeclaredColumnsAndCatalogTableIsSealed() + { + // A sealed table declares everything, so a segment carrying a column the spec does not have is a real difference + // and must be recompacted away. + final DataSourceCompactionConfig sealed = catalogCompactionConfig(SEALED_WIKI); + final DataSegment segment = DataSegment + .builder(WIKI_SEGMENT) + .lastCompactionState(stateWithUndeclaredColumn(sealed)) + .build(); + + final CompactionStatus status = CompactionStatus.compute(List.of(segment), sealed, fingerprintMapper); + Assertions.assertFalse(status.isComplete()); + Assertions.assertTrue(status.getReason().contains("baseTable")); + } + + @Test + public void testStatusWhenBaseTableDeclaredColumnChangesAndCatalogTableIsNotSealed() + { + // The relaxation only forgives columns appended after the declared ones; a change to the declaration itself is + // still a change, whether or not the table is sealed. + final DataSourceCompactionConfig unsealed = catalogCompactionConfig(UNSEALED_WIKI); + final CompactionState declared = unsealed.toCompactionState(); + final ClusteredValueGroupsBaseTableProjectionSpec reclustered = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant", "region") + .build(); + final DataSegment segment = DataSegment + .builder(WIKI_SEGMENT) + .lastCompactionState( + CompactionState.builder() + .partitionsSpec(declared.getPartitionsSpec()) + .indexSpec(declared.getIndexSpec()) + .segmentGranularitySpec(declared.getSegmentGranularitySpec()) + .baseTable(reclustered) + .build() + ) + .build(); + + final CompactionStatus status = CompactionStatus.compute(List.of(segment), unsealed, fingerprintMapper); + Assertions.assertFalse(status.isComplete()); + Assertions.assertTrue(status.getReason().contains("baseTable")); + } + + /** + * The state a compaction of {@code config} would record if the segments carried an extra column the catalog does not + * declare: everything the config asks for, plus the discovered column appended to the base table. + */ + private static CompactionState stateWithUndeclaredColumn(DataSourceCompactionConfig config) + { + final CompactionState declared = config.toCompactionState(); + return CompactionState.builder() + .partitionsSpec(declared.getPartitionsSpec()) + .indexSpec(declared.getIndexSpec()) + .segmentGranularitySpec(declared.getSegmentGranularitySpec()) + .baseTable( + declared.getBaseTable() + .withAdditionalColumns(List.of(new StringDimensionSchema("city"))) + ) + .build(); + } + + private static CatalogDataSourceCompactionConfig catalogCompactionConfig(String dataSource) + { + return new CatalogDataSourceCompactionConfig( + dataSource, + null, + null, + null, + null, + null, + null, + CATALOG + ); + } + @Test public void testStatusWhenProjectionsMismatch() { diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java index af779a7b840c..fdeb473bf669 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfigTest.java @@ -26,13 +26,21 @@ import nl.jqno.equalsverifier.EqualsVerifier; import org.apache.druid.catalog.MapMetadataCatalog; import org.apache.druid.catalog.MetadataCatalog; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; import org.apache.druid.catalog.model.ColumnSpec; import org.apache.druid.catalog.model.DatasourceProjectionMetadata; import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.table.ClusterKeySpec; import org.apache.druid.catalog.model.table.DatasourceDefn; import org.apache.druid.catalog.model.table.TableBuilder; import org.apache.druid.data.input.impl.AggregateProjectionSpec; +import org.apache.druid.data.input.impl.BaseTableProjectionSpec; +import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec; +import org.apache.druid.data.input.impl.DimensionSchema; import org.apache.druid.data.input.impl.StringDimensionSchema; +import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; +import org.apache.druid.indexer.partitions.PartitionsSpec; import org.apache.druid.jackson.DefaultObjectMapper; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.granularity.Granularities; @@ -47,11 +55,23 @@ import org.junit.jupiter.api.Test; import java.util.List; +import java.util.stream.Collectors; public class CatalogDataSourceCompactionConfigTest { private static final String TEST_DS = "test"; private static final String TEST_DS_PROJECTIONS_ONLY_SCHEMA = "test_projections"; + private static final String TEST_DS_CLUSTERED = "test_clustered"; + private static final String TEST_DS_CLUSTERED_UNSEALED = "test_clustered_unsealed"; + private static final String TEST_DS_NO_SEGMENT_GRANULARITY = "test_no_segment_granularity"; + private static final String TEST_DS_CLUSTER_KEYS = "test_cluster_keys"; + private static final String TEST_DS_CLUSTERED_WITH_KEYS = "test_clustered_with_keys"; + + private static final List CLUSTERED_COLUMNS = ImmutableList.of( + new ColumnSpec("tenant", ColumnType.STRING.asTypeString(), null), + new ColumnSpec(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG.asTypeString(), null), + new ColumnSpec("string", ColumnType.STRING.asTypeString(), null) + ); private static final ObjectMapper MAPPER; private static final MapMetadataCatalog METADATA_CATALOG; @@ -109,6 +129,162 @@ public class CatalogDataSourceCompactionConfigTest .buildSpec() ); + METADATA_CATALOG.addSpec( + TableId.datasource(TEST_DS_CLUSTERED), + TableBuilder.datasource(TEST_DS_CLUSTERED, "P1D") + .columns(CLUSTERED_COLUMNS) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("tenant"), null, null)) + .sealed(true) + .buildSpec() + ); + METADATA_CATALOG.addSpec( + TableId.datasource(TEST_DS_CLUSTERED_UNSEALED), + TableBuilder.datasource(TEST_DS_CLUSTERED_UNSEALED, "P1D") + .columns(CLUSTERED_COLUMNS) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("tenant"), null, null)) + .buildSpec() + ); + METADATA_CATALOG.addSpec( + TableId.datasource(TEST_DS_CLUSTER_KEYS), + TableBuilder.datasource(TEST_DS_CLUSTER_KEYS, "P1D") + .columns( + ImmutableList.of( + new ColumnSpec(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG.asTypeString(), null), + new ColumnSpec("string", ColumnType.STRING.asTypeString(), null), + new ColumnSpec("double", ColumnType.DOUBLE.asTypeString(), null) + ) + ) + .property(DatasourceDefn.TARGET_SEGMENT_ROWS_PROPERTY, 1000) + .clusterColumns(new ClusterKeySpec("string", false), new ClusterKeySpec("double", false)) + .buildSpec() + ); + METADATA_CATALOG.addSpec( + TableId.datasource(TEST_DS_CLUSTERED_WITH_KEYS), + TableBuilder.datasource(TEST_DS_CLUSTERED_WITH_KEYS, "P1D") + .columns(CLUSTERED_COLUMNS) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("tenant"), null, null)) + .clusterColumns(new ClusterKeySpec("string", false)) + .buildSpec() + ); + METADATA_CATALOG.addSpec( + TableId.datasource(TEST_DS_NO_SEGMENT_GRANULARITY), + TableBuilder.datasource(TEST_DS_NO_SEGMENT_GRANULARITY, "P1D") + .columns(CLUSTERED_COLUMNS) + .property(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY, null) + .buildSpec() + ); + } + + private static CatalogDataSourceCompactionConfig configFor(String dataSource) + { + return new CatalogDataSourceCompactionConfig( + dataSource, + null, + null, + null, + null, + null, + null, + METADATA_CATALOG + ); + } + + @Test + public void testBaseTableIsDerivedFromCatalogLayoutAndColumns() + { + // The layout carries only what the column list cannot express (the clustering prefix); the declared columns, + // in their declared order, are the schema. + final BaseTableProjectionSpec baseTable = configFor(TEST_DS_CLUSTERED).getBaseTable(); + + Assertions.assertInstanceOf(ClusteredValueGroupsBaseTableProjectionSpec.class, baseTable); + Assertions.assertEquals( + ImmutableList.of("tenant", ColumnHolder.TIME_COLUMN_NAME, "string"), + baseTable.getDimensionsSpec() + .getDimensions() + .stream() + .map(DimensionSchema::getName) + .collect(Collectors.toList()) + ); + Assertions.assertEquals( + ImmutableList.of("tenant"), + ((ClusteredValueGroupsBaseTableProjectionSpec) baseTable).getClusteringColumnNames() + ); + } + + @Test + public void testBaseTableIsNullWhenTableDeclaresNoLayout() + { + Assertions.assertNull(configFor(TEST_DS).getBaseTable()); + } + + @Test + public void testBaseTableIsNullWhenTableIsNotInCatalog() + { + Assertions.assertNull(configFor("not_in_catalog").getBaseTable()); + } + + @Test + public void testIsSealedFollowsCatalogProperty() + { + Assertions.assertTrue(configFor(TEST_DS_CLUSTERED).isSealed()); + // An absent 'sealed' property means the table accepts columns it does not declare, so compaction must analyze the + // existing segments to find and preserve them. + Assertions.assertFalse(configFor(TEST_DS_CLUSTERED_UNSEALED).isSealed()); + Assertions.assertFalse(configFor(TEST_DS).isSealed()); + // No table at all: there is no baseTable either, so nothing observes this; report the safe default. + Assertions.assertTrue(configFor("not_in_catalog").isSealed()); + } + + @Test + public void testSegmentGranularityIsNullWhenTableDeclaresNone() + { + // The coordinator asks for this on every evaluation, so a table with no segmentGranularity must not blow up. + Assertions.assertNull(configFor(TEST_DS_NO_SEGMENT_GRANULARITY).getSegmentGranularity()); + Assertions.assertNull(configFor(TEST_DS_NO_SEGMENT_GRANULARITY).getGranularitySpec().getSegmentGranularity()); + Assertions.assertNull(configFor("not_in_catalog").getSegmentGranularity()); + } + + @Test + public void testTuningConfigPartitionsByClusterKeys() + { + // Cluster keys are the table's statement of how rows should be laid out, so compaction range-partitions by them + // rather than falling back to dynamic partitioning. + final PartitionsSpec partitionsSpec = configFor(TEST_DS_CLUSTER_KEYS).getTuningConfig().getPartitionsSpec(); + + Assertions.assertInstanceOf(DimensionRangePartitionsSpec.class, partitionsSpec); + Assertions.assertEquals( + ImmutableList.of("string", "double"), + ((DimensionRangePartitionsSpec) partitionsSpec).getPartitionDimensions() + ); + // targetSegmentRows is the segment size, carried as maxRowsPerSegment so it is used verbatim rather than scaled + // by 1.5x the way a targetRowsPerSegment would be -- an INSERT into this table sizes segments the same way. + Assertions.assertNull(((DimensionRangePartitionsSpec) partitionsSpec).getTargetRowsPerSegment()); + Assertions.assertEquals(1000, partitionsSpec.getMaxRowsPerSegment()); + } + + @Test + public void testTuningConfigPartitionsByClusterKeysAlongsideBaseTable() + { + // A base table's clustering columns group rows within a segment; cluster keys split rows across segments. A table + // that declares both gets both. + final PartitionsSpec partitionsSpec = + configFor(TEST_DS_CLUSTERED_WITH_KEYS).getTuningConfig().getPartitionsSpec(); + + Assertions.assertEquals( + ImmutableList.of("string"), + ((DimensionRangePartitionsSpec) partitionsSpec).getPartitionDimensions() + ); + Assertions.assertNotNull(configFor(TEST_DS_CLUSTERED_WITH_KEYS).getBaseTable()); + } + + @Test + public void testTuningConfigFallsBackToDynamicPartitioning() + { + final PartitionsSpec partitionsSpec = configFor(TEST_DS).getTuningConfig().getPartitionsSpec(); + + Assertions.assertInstanceOf(DynamicPartitionsSpec.class, partitionsSpec); + Assertions.assertNull(((DynamicPartitionsSpec) partitionsSpec).getMaxTotalRows()); + Assertions.assertNull(configFor("not_in_catalog").getTuningConfig()); } @Test diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java index 2c50fc4ec7d6..08da7232dbe2 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java @@ -31,6 +31,8 @@ import org.apache.druid.catalog.MapMetadataCatalog; import org.apache.druid.catalog.MetadataCatalog; import org.apache.druid.catalog.NullMetadataCatalog; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; import org.apache.druid.catalog.model.DatasourceProjectionMetadata; import org.apache.druid.catalog.model.TableId; import org.apache.druid.catalog.model.table.DatasourceDefn; @@ -79,6 +81,8 @@ import org.apache.druid.rpc.indexing.NoopOverlordClient; import org.apache.druid.rpc.indexing.OverlordClient; import org.apache.druid.segment.IndexSpec; +import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.incremental.OnheapIncrementalIndex; import org.apache.druid.segment.indexing.BatchIOConfig; import org.apache.druid.segment.transform.CompactionTransformSpec; @@ -1066,6 +1070,68 @@ public void testCompactWithCatalogProjections() ); } + @Test + public void testCompactWithCatalogBaseTable() + { + // The catalog table's layout combines with its declared columns to become the physical spec the compaction task + // is given, and the table's 'sealed' property decides whether the task may treat that spec as complete. + final String dataSource = DATA_SOURCE_PREFIX + 0; + final OverlordClient mockClient = Mockito.mock(OverlordClient.class); + final ObjectMapper mapper = new DefaultObjectMapper((DefaultObjectMapper) JSON_MAPPER); + final MapMetadataCatalog metadataCatalog = new MapMetadataCatalog(mapper); + + mapper.setInjectableValues( + new InjectableValues.Std() + .addValue(MetadataCatalog.class, metadataCatalog) + .addValue(ExprMacroTable.class, TestExprMacroTable.INSTANCE) + ); + + final ArgumentCaptor payloadCaptor = setUpMockClient(mockClient); + metadataCatalog.addSpec( + TableId.datasource(dataSource), + TableBuilder.datasource(dataSource, "P1D") + .columns( + ImmutableList.of( + new ColumnSpec("bar", ColumnType.STRING.asTypeString(), null), + new ColumnSpec(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG.asTypeString(), null) + ) + ) + .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("bar"), null, null)) + .buildSpec() + ); + final CompactSegments compactSegments = new CompactSegments(statusTracker, mockClient); + doCompactSegments( + compactSegments, + List.of( + new CatalogDataSourceCompactionConfig( + dataSource, + // baseTable requires the MSQ compaction engine + CompactionEngine.MSQ, + new Period("PT0H"), + null, + 0, + null, + 500L, + metadataCatalog + ) + ) + ); + + final ClientCompactionTaskQuery taskPayload = (ClientCompactionTaskQuery) payloadCaptor.getValue(); + Assertions.assertEquals( + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .columns( + new StringDimensionSchema("bar"), + new LongDimensionSchema(ColumnHolder.TIME_COLUMN_NAME) + ) + .clusteringColumns("bar") + .build(), + taskPayload.getBaseTable() + ); + // The table does not declare itself sealed, so the task is told to analyze the segments it is rewriting. + Assertions.assertFalse(taskPayload.isSealed()); + } + @Test public void testCompactWithRollupInGranularitySpec() { @@ -1144,6 +1210,7 @@ public void testCompactWithGranularitySpecConflictWithActiveCompactionTask() null, null, null, + null, null ) );