capistrant commented on code in PR #20379:
URL: https://github.com/apache/druid/pull/20379#discussion_r4327465695
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java:
##########
@@ -797,16 +844,69 @@ private static DataSchema createDataSchema(
@Nonnull ClientCompactionTaskGranularitySpec granularitySpec,
@Nullable List<AggregateProjectionSpec> projections,
@Nullable BaseTableProjectionSpec baseTable,
+ boolean sealed,
+ boolean rangePartitioned,
boolean needMultiValuedColumns
)
{
if (baseTable != null) {
// 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());
+ // granularity; segment granularity + intervals are the compaction
config's. A sealed spec is the complete
+ // description of the table, so the columns and projections not declared
are taken to not exist (and will be
+ // removed).
+ //
+ // Which string columns hold multiple values is only knowable from the
segments, so a sealed table declares the
+ // answer instead of paying to read it: a declared string column holds
one value per row, since a base table has
+ // no way to say otherwise yet. A table whose data disagrees fails
rather than being stored wrongly, after the
+ // shuffle for a range partition column and at the groupBy for a rollup
grouping column.
+ final boolean analyzeSchema = !sealed;
+ final boolean analyzeProjections = !sealed && projections == null;
+ final boolean analyzeMultiValued = analyzeSchema && (rangePartitioned ||
needMultiValuedColumns);
+ BaseTableProjectionSpec baseTableSpec =
baseTable.withQueryGranularity(granularitySpec.getQueryGranularity());
+ List<AggregateProjectionSpec> projectionSpecs = projections;
+ Set<String> multiValuedDimensions = sealed ? Set.of() : null;
+ if (analyzeSchema || analyzeMultiValued) {
+ final ExistingSegmentAnalyzer analyzer = new ExistingSegmentAnalyzer(
+ segments,
+ false,
+ false,
+ analyzeSchema,
+ analyzeSchema,
+ analyzeProjections,
+ analyzeMultiValued
+ );
+ fetchAndProcess(analyzer, emitter, metricBuilder);
+ if (analyzeSchema) {
+ // An unsealed spec is partial, so append what it does not declare
after the declared columns, leaving the
+ // shape the operator asked for unchanged.
+ baseTableSpec = baseTableSpec.withAdditionalColumns(
+ undeclaredColumns(baseTableSpec, analyzer.getDimensionsSpec())
+ );
+ if (baseTableSpec instanceof RollupBaseTableProjectionSpec) {
+ baseTableSpec = ((RollupBaseTableProjectionSpec)
baseTableSpec).withAdditionalMetrics(
+ undeclaredMetrics(baseTableSpec, analyzer.getMetricsSpec())
Review Comment:
altering the type of a declared metric could break compaction cuz
getMetricsSpec would start throwing on segments who have different types but
same name. first thought is to allow discarding declared metrics before merging
what the analyzer found?
1. longSum(total) used for a bit in catalog.
2. operator changes to doubleSum(total) in catalog
3. compaction runs over a day of segments that saw both --> I think it would
blow up. but we would have rather it compacted towards doubleSum as declared.
##########
server/src/main/java/org/apache/druid/server/coordinator/CatalogDataSourceCompactionConfig.java:
##########
@@ -238,9 +336,53 @@ public AggregatorFactory[] getMetricsSpec()
@Override
public CompactionConfigValidationResult validate(ClusterCompactionConfig
clusterCompactionConfig)
{
+ final CompactionEngine effectiveEngine = engine == null ?
clusterCompactionConfig.getEngine() : engine;
+ if (effectiveEngine == CompactionEngine.MSQ) {
+ // the validation below can only check partition column types when the
config has a schema to read them from,
+ // which a catalog table without a baseTable layout does not, so check
the declared types here instead
+ final CompactionConfigValidationResult clusterKeyTypes =
validateClusterKeyTypes();
+ if (!clusterKeyTypes.isValid()) {
+ return clusterKeyTypes;
+ }
+ }
return ClientCompactionRunnerInfo.validateCompactionConfig(this,
clusterCompactionConfig.getEngine());
}
+ /**
+ * Checks that every cluster key names a declared {@code VARCHAR} column,
the only type the MSQ engine range
+ * partitions by. A key naming a column the table does not declare is left
alone: the table may still store it, and
+ * its type is not knowable from the catalog.
+ */
+ private CompactionConfigValidationResult validateClusterKeyTypes()
+ {
+ final ResolvedTable table = catalog.resolveTable(tableId);
+ final List<ClusterKeySpec> clusterKeys = getClusterKeys(table);
+ if (CollectionUtils.isNullOrEmpty(clusterKeys)) {
+ return CompactionConfigValidationResult.success();
+ }
+ final List<ColumnSpec> columns = table.spec().columns();
+ if (CollectionUtils.isNullOrEmpty(columns)) {
+ return CompactionConfigValidationResult.success();
+ }
+ for (ClusterKeySpec clusterKey : clusterKeys) {
+ final int position = CatalogUtils.findColumn(columns, clusterKey.expr());
+ if (position < 0) {
+ continue;
+ }
+ final ColumnType type = Columns.druidType(columns.get(position));
+ if (!ColumnType.STRING.equals(type)) {
Review Comment:
what do you think about a special handling failure for untyped catalog
columns to try and push an operator to have typed columns everywhere? I believe
DDL bans untyped columns but pre-existing tables and API created tables might
still allow that path?
```
"Cluster key[%s] of table[%s] is declared without a type. Compaction
range-partitions by this column, which the MSQ engine supports for VARCHAR
columns only. Declare the column as VARCHAR, or choose a different cluster key."
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java:
##########
@@ -797,16 +844,69 @@ private static DataSchema createDataSchema(
@Nonnull ClientCompactionTaskGranularitySpec granularitySpec,
@Nullable List<AggregateProjectionSpec> projections,
@Nullable BaseTableProjectionSpec baseTable,
+ boolean sealed,
+ boolean rangePartitioned,
boolean needMultiValuedColumns
)
{
if (baseTable != null) {
// 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());
+ // granularity; segment granularity + intervals are the compaction
config's. A sealed spec is the complete
+ // description of the table, so the columns and projections not declared
are taken to not exist (and will be
+ // removed).
+ //
+ // Which string columns hold multiple values is only knowable from the
segments, so a sealed table declares the
+ // answer instead of paying to read it: a declared string column holds
one value per row, since a base table has
+ // no way to say otherwise yet. A table whose data disagrees fails
rather than being stored wrongly, after the
+ // shuffle for a range partition column and at the groupBy for a rollup
grouping column.
+ final boolean analyzeSchema = !sealed;
+ final boolean analyzeProjections = !sealed && projections == null;
+ final boolean analyzeMultiValued = analyzeSchema && (rangePartitioned ||
needMultiValuedColumns);
+ BaseTableProjectionSpec baseTableSpec =
baseTable.withQueryGranularity(granularitySpec.getQueryGranularity());
+ List<AggregateProjectionSpec> projectionSpecs = projections;
+ Set<String> multiValuedDimensions = sealed ? Set.of() : null;
+ if (analyzeSchema || analyzeMultiValued) {
+ final ExistingSegmentAnalyzer analyzer = new ExistingSegmentAnalyzer(
+ segments,
+ false,
+ false,
+ analyzeSchema,
+ analyzeSchema,
+ analyzeProjections,
+ analyzeMultiValued
+ );
+ fetchAndProcess(analyzer, emitter, metricBuilder);
+ if (analyzeSchema) {
+ // An unsealed spec is partial, so append what it does not declare
after the declared columns, leaving the
+ // shape the operator asked for unchanged.
+ baseTableSpec = baseTableSpec.withAdditionalColumns(
+ undeclaredColumns(baseTableSpec, analyzer.getDimensionsSpec())
+ );
+ if (baseTableSpec instanceof RollupBaseTableProjectionSpec) {
+ baseTableSpec = ((RollupBaseTableProjectionSpec)
baseTableSpec).withAdditionalMetrics(
+ undeclaredMetrics(baseTableSpec, analyzer.getMetricsSpec())
+ );
+ } else {
+ // A spec with no metric columns has nowhere to put these, and
dropping them would delete data the table
+ // never said it did not want. Sealing the table is how an
operator asks for that.
+ final Set<String> metricNames = analyzer.getMetricNames();
Review Comment:
unsealed non rollup tables whose compaction candidate data has a metric with
a name that matches a declared column leads to misleading error output.
I think calling out the conflict between declared columns and discovered
metrics could help an operator decide what to do.
```
Segments of dataSource[%s] store %s as metric column(s) that the table
declares as regular column(s). An unsealed non-rollup base table cannot compact
these. Seal the table to compact them into the declared columns (sealing also
drops any undeclared column), or use a rollup base table to keep them as
metrics.
```
or we could treat metric --> declared column for unsealed as allowed and
automatic? like sealed does?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
