feat: allow non-sealed schemas for MSQ when using base table spec

multi-stage-query/src/main/java/org/apache/druid/msq/indexing/destination/SegmentGenerationUtils.java

@@ -28,6 +28,7 @@
 import org.apache.druid.data.input.impl.LongDimensionSchema;
 import org.apache.druid.data.input.impl.TimestampSpec;
 import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.frame.key.ClusterBy;
 import org.apache.druid.frame.key.KeyColumn;
 import org.apache.druid.indexer.granularity.ArbitraryGranularitySpec;
@@ -66,6 +67,7 @@
 import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -104,8 +106,14 @@ public static DataSchema makeDataSchemaForIngestion(
     if (destination.getBaseTable() != null) {
       final Granularity queryGranularity =
           query.context().getGranularity(DruidSqlInsert.SQL_INSERT_QUERY_GRANULARITY, jsonMapper);
-      final BaseTableProjectionSpec baseTable =
+      final BaseTableProjectionSpec declared =
           destination.getBaseTable().withQueryGranularity(queryGranularity);
+      // The query may produce columns the base table does not declare, which happens when the target table does not
+      // require a strict schema. Store them rather than drop them; they are appended after the declared columns, so
+      // the shape the operator asked for is unchanged.
+      final BaseTableProjectionSpec baseTable = declared.withAdditionalColumns(
+          undeclaredColumns(declared, querySignature, columnMappings, query, destination.getDimensionSchemas())
+      );
       return DataSchema.builder()
                        .withDataSource(destination.getDataSource())
                        .withTimestamp(new TimestampSpec(ColumnHolder.TIME_COLUMN_NAME, "millis", null))
@@ -212,6 +220,59 @@ private static boolean isRollupQuery(Query<?> query)
            && !query.context().getBoolean(GroupByQueryConfig.CTX_KEY_ENABLE_MULTI_VALUE_UNNESTING, true);
   }
 
+  /**
+   * The columns a query produces that a base table does not declare, in query output order, as the
+   * {@link DimensionSchema}s they should be stored with.
+   * <p>
+   * A base table declares the columns an operator asked for. When the target table does not require a strict schema,
+   * a query may produce others; they are stored so that ingesting a column is never silently a no-op.
+   */
+  private static List<DimensionSchema> undeclaredColumns(
+      final BaseTableProjectionSpec baseTable,
+      final RowSignature querySignature,
+      final ColumnMappings columnMappings,
+      final Query<?> query,
+      @Nullable final Map<String, DimensionSchema> dimensionSchemas
+  )
+  {
+    final Set<String> declared = new HashSet<>();
+    for (DimensionSchema dimension : baseTable.getDimensionsSpec().getDimensions()) {
+      declared.add(dimension.getName());
+    }
+    for (AggregatorFactory metric : baseTable.getMetrics() == null ? new AggregatorFactory[0] : baseTable.getMetrics()) {
+      declared.add(metric.getName());
+    }
+    // The time column is positional in a base table, never appended.
+    declared.add(ColumnHolder.TIME_COLUMN_NAME);
+
+    final List<DimensionSchema> undeclared = new ArrayList<>();
+    for (final String outputColumnName : columnMappings.getOutputColumnNames()) {
+      if (!declared.add(outputColumnName)) {
+        continue;
+      }
+      final int outputColumn = CollectionUtils.getOnlyElement(
+          columnMappings.getOutputColumnsByName(outputColumnName),
+          xs -> new ISE("Expected single output column for name [%s], but got [%s]", outputColumnName, xs)
+      );
+      final String queryColumn = columnMappings.getQueryColumnName(outputColumn);
+      final ColumnType type =
+          querySignature.getColumnType(queryColumn)
+                        .orElseThrow(() -> new ISE("No type for column [%s]", outputColumnName));
+
+      if (type.is(ValueType.COMPLEX) && !DimensionHandlerUtils.DIMENSION_HANDLER_PROVIDERS.containsKey(type.getComplexTypeName())) {
+        // A base table stores columns as dimensions, so a complex type with no dimension handler has nowhere to go
+        throw InvalidInput.exception(
+            "Column [%s] has type [%s], which cannot be stored in a base table that does not declare it. Declare the"
+            + " column in the table, or cast it to a type that can be stored as a dimension",
+            outputColumnName,
+            type
+        );
+      }
+      undeclared.add(getDimensionSchema(outputColumnName, type, query.context(), dimensionSchemas));
+    }
+    return undeclared;
+  }
+
   private static DimensionSchema getDimensionSchema(
       final String outputColumnName,
       @Nullable final ColumnType queryType,

multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java

@@ -281,6 +281,17 @@
                     )
                     .buildSpec()
     );
+    metadataCatalog.addSpec(
+        TableId.datasource("fooClusteredUnsealed"),
+        // Not sealed: the table declares the layout it cares about (clustering column, time position, delta) and lets
+        // a query bring whatever else it produces.
+        TableBuilder.datasource("fooClusteredUnsealed", Granularities.DAY.toString())
+                    .column("channel", Columns.SQL_VARCHAR)
+                    .timeColumn()
+                    .column("delta", Columns.SQL_BIGINT)
+                    .baseTable(new ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("channel"), null, null))
+                    .buildSpec()
+    );
     return new LiveCatalogResolver(metadataCatalog);
   }
 
@@ -879,6 +890,128 @@
                      .verifyResults();
   }
 
+  @MethodSource("data")
+  @ParameterizedTest(name = "{index}:with context {0}")
+  public void testInsertOnExternalDataSourceWithUnsealedCatalogClusteredBaseTable(
+      String contextName,
+      Map<String, Object> context
+  ) throws IOException
+  {
+    final File toRead = getResourceAsTemporaryFile("/wikipedia-sampled.json");
+    final String toReadFileNameAsJson = queryFramework().queryJsonMapper().writeValueAsString(toRead.getAbsolutePath());
+
+    // The table declares only channel, __time and delta. The other four columns the query produces are appended after
+    // them in query output order, so the declared layout (clustering prefix, time position) is unchanged and nothing
+    // the query selected is dropped.
+    RowSignature rowSignature = RowSignature.builder()
+                                            .add("channel", ColumnType.STRING)
+                                            .add("__time", ColumnType.LONG)
+                                            .add("delta", ColumnType.LONG)
+                                            .add("page", ColumnType.STRING)
+                                            .add("user", ColumnType.STRING)
+                                            .add("added", ColumnType.LONG)
+                                            .add("deleted", ColumnType.LONG)
+                                            .build();
+
+    testIngestQuery().setSql(" insert into fooClusteredUnsealed SELECT\n"
+                             + "  floor(TIME_PARSE(\"timestamp\") to minute) AS __time,\n"
+                             + "  channel,\n"
+                             + "  page,\n"
+                             + "  user,\n"
+                             + "  added,\n"
+                             + "  deleted,\n"
+                             + "  delta\n"
+                             + "FROM TABLE(\n"
+                             + "  EXTERN(\n"
+                             + "    '{ \"files\": [" + toReadFileNameAsJson + "],\"type\":\"local\"}',\n"
+                             + "    '{\"type\": \"json\"}',\n"
+                             + "    '[{\"name\": \"timestamp\", \"type\": \"string\"}, {\"name\": \"channel\", \"type\": \"string\"}, {\"name\": \"page\", \"type\": \"string\"}, {\"name\": \"user\", \"type\": \"string\"}, {\"name\": \"added\", \"type\": \"long\"}, {\"name\": \"deleted\", \"type\": \"long\"}, {\"name\": \"delta\", \"type\": \"long\"}]'\n"
+                             + "  )\n"
+                             + ") PARTITIONED by day ")
+                     .setExpectedDataSource("fooClusteredUnsealed")
+                     .setExpectedRowSignature(rowSignature)
+                     .setQueryContext(context)
+                     .setExpectedSegments(ImmutableSet.of(SegmentId.of(
+                         "fooClusteredUnsealed",
+                         Intervals.of("2016-06-27/P1D"),
+                         "test",
+                         0
+                     )))
+                     // The appended columns are not clustered on; clustering is what the table declared.
+                     .setExpectedClusterGroups(
+                         new ClusterGroupTuples(
+                             RowSignature.builder().add("channel", ColumnType.STRING).build(),
+                             ImmutableList.of(
+                                 ImmutableList.of("#ceb.wikipedia"),
+                                 ImmutableList.of("#de.wikipedia"),
+                                 ImmutableList.of("#en.wikipedia"),
+                                 ImmutableList.of("#es.wikipedia"),
+                                 ImmutableList.of("#id.wikipedia"),
+                                 ImmutableList.of("#pl.wikipedia"),
+                                 ImmutableList.of("#pt.wikipedia"),
+                                 ImmutableList.of("#ru.wikipedia"),
+                                 ImmutableList.of("#sh.wikipedia"),
+                                 ImmutableList.of("#sv.wikipedia"),
+                                 ImmutableList.of("#zh.wikipedia")
+                             )
+                         )
+                     )
+                     // Rows are read back in segment order, which now sorts by delta ahead of the appended columns.
+                     .setExpectedResultRows(
+                         ImmutableList.of(
+                             new Object[]{"#ceb.wikipedia", 1466985660000L, 4150L, "Neqerssuaq", "Lsjbot", 4150L, 0L},
+                             new Object[]{"#de.wikipedia", 1466992920000L, 2560L, "Benutzer Diskussion:Squasher/Archiv/2016", "TaxonBot", 2560L, 0L},
+                             new Object[]{"#de.wikipedia", 1466992980000L, 364L, "Benutzer Diskussion:HerrSonderbar", "GiftBot", 364L, 0L},
+                             new Object[]{"#en.wikipedia", 1466985600000L, -2L, "Richie Rich's Christmas Wish", "JasonAQuest", 0L, 2L},
+                             new Object[]{"#en.wikipedia", 1466985600000L, 2L, "Bailando 2015", "181.230.118.178", 2L, 0L},
+                             new Object[]{"#en.wikipedia", 1466985660000L, 496L, "Panama Canal", "Mariordo", 496L, 0L},
+                             new Object[]{"#en.wikipedia", 1466992980000L, -463L, "File:Paint.net 4.0.6 screenshot.png", "Calvin Hogg", 0L, 463L},
+                             new Object[]{"#es.wikipedia", 1466985660000L, -173L, "Sumo (banda)", "181.110.165.189", 0L, 173L},
+                             new Object[]{"#es.wikipedia", 1466989320000L, 4L, "Clasificación para la Eurocopa Sub-21 de 2017", "Guly600", 4L, 0L},
+                             new Object[]{"#id.wikipedia", 1466989320000L, 106L, "Ibnu Sina", "Ftihikam", 106L, 0L},
+                             new Object[]{"#pl.wikipedia", 1466985600000L, 270L, "Kategoria:Dyskusje nad usunięciem artykułu zakończone bez konsensusu − lipiec 2016", "Beau.bot", 270L, 0L},
+                             new Object[]{"#pt.wikipedia", 1466992920000L, 1926L, "Dobromir Zhechev", "Ceresta", 1926L, 0L},
+                             new Object[]{"#ru.wikipedia", 1466985720000L, 196L, "Википедия:Опросы/Унификация шаблонов «Не переведено»", "Wanderer777", 196L, 0L},
+                             new Object[]{"#sh.wikipedia", 1466985660000L, -1L, "El Terco, Bachíniva", "Kolega2357", 0L, 1L},
+                             new Object[]{"#sh.wikipedia", 1466985720000L, -1L, "Hermanos Díaz, Ascensión", "Kolega2357", 0L, 1L},
+                             new Object[]{"#sh.wikipedia", 1466989320000L, -1L, "El Sicomoro, Ascensión", "Kolega2357", 0L, 1L},
+                             new Object[]{"#sh.wikipedia", 1466992920000L, -1L, "Trinidad Jiménez G., Benemérito de las Américas", "Kolega2357", 0L, 1L},
+                             new Object[]{"#sv.wikipedia", 1466985600000L, 31L, "Salo Toraut", "Lsjbot", 31L, 0L},
+                             new Object[]{"#zh.wikipedia", 1466989320000L, 18L, "中共十八大以来的反腐败工作", "2001:DA8:207:E132:94DC:BA03:DFDF:8F9F", 18L, 0L},
+                             new Object[]{"#zh.wikipedia", 1466992920000L, 1986L, "Wikipedia:頁面存廢討論/記錄/2016/06/27", "Tigerzeng", 1986L, 0L}
+                         )
+                     )
+                     .verifyResults();
+  }
+
+  @MethodSource("data")
+  @ParameterizedTest(name = "{index}:with context {0}")
+  public void testInsertOnUnsealedCatalogClusteredBaseTableUnstorableColumn(
+      String contextName,
+      Map<String, Object> context
+  )
+  {
+    // A base table stores columns as dimensions, so an undeclared sketch column has nowhere to go: it could only be
+    // stored as a metric, and a base table declares its own metrics.
+    testIngestQuery().setSql(
+                         "insert into fooClusteredUnsealed "
+                         + "select __time, dim1 as channel, cnt as delta, unique_dim1 as unique_users "
+                         + "from foo partitioned by day"
+                     )
+                     .setQueryContext(context)
+                     .setExpectedExecutionErrorMatcher(
+                         CoreMatchers.allOf(
+                             CoreMatchers.instanceOf(ISE.class),
+                             ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString(
+                                 "Column [unique_users] has type [COMPLEX<hyperUnique>], which cannot be stored in a"
+                                 + " base table that does not declare it. Declare the column in the table, or cast it"
+                                 + " to a type that can be stored as a dimension"
+                             ))
+                         )
+                     )
+                     .verifyExecutionError();
+  }
+
   @MethodSource("data")
   @ParameterizedTest(name = "{index}:with context {0}")
   public void testInsertOnFoo1WithGroupByLimitWithoutClusterBy(String contextName, Map<String, Object> context)

processing/src/main/java/org/apache/druid/data/input/impl/AdaptedBaseTableProjectionSpec.java

@@ -26,6 +26,7 @@
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.segment.VirtualColumns;
 import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.utils.CollectionUtils;
 
 import javax.annotation.Nullable;
 import java.util.ArrayList;
@@ -70,6 +71,17 @@ public GranularitySpec getGranularitySpec()
     return granularitySpec;
   }
 
+  @Override
+  public AdaptedBaseTableProjectionSpec withAdditionalColumns(@Nullable List<DimensionSchema> additionalColumns)
+  {
+    if (CollectionUtils.isNullOrEmpty(additionalColumns)) {
+      return this;
+    }
+    final List<DimensionSchema> revised = new ArrayList<>(dimensionsSpec.getDimensions());
+    revised.addAll(additionalColumns);
+    return new AdaptedBaseTableProjectionSpec(granularitySpec, dimensionsSpec.withDimensions(revised), metrics);
+  }
+
   @Override
   public VirtualColumns getVirtualColumns()
   {

processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java

@@ -85,6 +85,11 @@ default Granularity getQueryGranularity()
    */
   BaseTableProjectionSpec withQueryGranularity(@Nullable Granularity queryGranularity);
 
+  /**
+   * Returns a copy of this spec with the given columns appended to those it already declares.
+   */
+  BaseTableProjectionSpec withAdditionalColumns(@Nullable List<DimensionSchema> additionalColumns);
+
   /**
    * Returns true if this spec is equivalent to {@code other} for the purpose of deciding whether a segment is already
    * compacted. Segment granularity, query granularity, and rollup are each compared by their own compaction check

processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java

@@ -229,6 +229,35 @@ public boolean hasEqualCompactionState(BaseTableProjectionSpec other)
         .equals(((ClusteredValueGroupsBaseTableProjectionSpec) other).withoutQueryGranularity());
   }
 
+  /**
+   * Appends the given columns after the declared ones, which keeps the clustering columns the leading prefix of
+   * {@link #getColumns()}.
+   */
+  @Override
+  public ClusteredValueGroupsBaseTableProjectionSpec withAdditionalColumns(
+      @Nullable List<DimensionSchema> additionalColumns
+  )
+  {
+    if (CollectionUtils.isNullOrEmpty(additionalColumns)) {
+      return this;
+    }
+    final List<DimensionSchema> revised = new ArrayList<>(columns.size() + additionalColumns.size());
+    revised.addAll(columns);
+    for (DimensionSchema additionalColumn : additionalColumns) {
+      if (ColumnHolder.TIME_COLUMN_NAME.equals(additionalColumn.getName())) {
+        throw InvalidInput.exception(
+            "Cannot append column [%s] to a [%s] base table; it must be declared at its position in the column list",
+            ColumnHolder.TIME_COLUMN_NAME,
+            TYPE_NAME
+        );
+      }
+      revised.add(additionalColumn);
+    }
+    // Duplicates of a declared column, and of a column materialized by a virtual column, are rejected by the
+    // constructor's validation.
+    return new ClusteredValueGroupsBaseTableProjectionSpec(virtualColumns, revised, clusteringColumns);
+  }
+
   /**
    * Returns a copy of this spec with the {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} virtual column removed,
    * the inverse of {@link #withQueryGranularity(Granularity)}. If no such virtual column is present this returns