feat: add plain table base table spec for catalog wiring

embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogDdlAndIngestTest.java

@@ -407,6 +407,102 @@ public void testCreateClusteredBaseTableThenIngestAndQuery()
     );
   }
 
+  /**
+   * A computed column composed of specialized expressions is lifted with its dependency closure (the inner JSON_MERGE
+   * becomes an intermediary virtual column the outer nested-field virtual column reads by synthetic name), and the
+   * clustered write path evaluates the chain at ingest to materialize the declared column.
+   */
+  @Test
+  public void testCreateClusteredBaseTableWithComposedComputedColumn()
+  {
+    final String tableName = dataSource;
+
+    cluster.callApi().runSql(
+        "CREATE TABLE \"%s\" (\n"
+        + "  tenant VARCHAR,\n"
+        + "  __time TIMESTAMP,\n"
+        + "  payload TYPE('COMPLEX<json>'),\n"
+        + "  v BIGINT,\n"
+        + "  PROJECTION __base AS (\n"
+        + "    SELECT tenant, __time, payload, JSON_VALUE(JSON_MERGE(payload, payload), '$.v' RETURNING BIGINT) AS v\n"
+        + "    CLUSTERED BY tenant\n"
+        + "  )\n"
+        + ")\n"
+        + "PARTITIONED BY DAY",
+        tableName
+    );
+
+    // 'v' is computed by the base table, so the INSERT supplies only its input column.
+    ingest(
+        "INSERT INTO \"%s\"\n"
+        + "SELECT TIME_PARSE(t) AS __time, a AS tenant, p AS payload\n"
+        + "FROM TABLE(\n"
+        + "  EXTERN(\n"
+        + "    '{\"type\":\"inline\",\"data\":\""
+        + "{\\\"t\\\":\\\"2022-12-26T12:34:56\\\",\\\"a\\\":\\\"bbb\\\",\\\"p\\\":{\\\"v\\\":7}}"
+        + "\\n{\\\"t\\\":\\\"2022-12-26T12:34:56\\\",\\\"a\\\":\\\"aaa\\\",\\\"p\\\":{\\\"v\\\":12}}\"}',\n"
+        + "    '{\"type\":\"json\"}'\n"
+        + "  )\n"
+        + ") EXTEND (t VARCHAR, a VARCHAR, p TYPE('COMPLEX<json>'))\n",
+        tableName
+    );
+
+    // Rows come back in clustering order, with 'v' materialized at ingest by the composed virtual column chain.
+    cluster.callApi().verifySqlQuery(
+        "SELECT tenant, v FROM %s",
+        tableName,
+        "aaa,12\n"
+        + "bbb,7"
+    );
+  }
+
+  /**
+   * A {@code __base} projection without {@code CLUSTERED BY} declares the plain-table layout: declared column order is
+   * the segment storage and sort order, and {@code TIME_FLOOR(__time, <period>)} declares the table's query
+   * granularity. Rows arrive unsorted with unfloored timestamps and come back sorted by the declared order (item
+   * before {@code __time}) with {@code __time} floored to the hour.
+   */
+  @Test
+  public void testCreatePlainBaseTableThenIngestAndQuery()
+  {
+    final String tableName = dataSource;
+
+    cluster.callApi().runSql(
+        "CREATE TABLE \"%s\" (\n"
+        + "  item VARCHAR,\n"
+        + "  __time TIMESTAMP,\n"
+        + "  val BIGINT,\n"
+        + "  PROJECTION __base AS (SELECT item, TIME_FLOOR(__time, 'PT1H') AS __time, val)\n"
+        + ")\n"
+        + "PARTITIONED BY DAY",
+        tableName
+    );
+
+    final TableMetadata table = client.readTable(TableId.datasource(tableName));
+    assertNotNull(table.spec().properties().get(DatasourceDefn.BASE_TABLE_PROPERTY));
+
+    ingest(
+        "INSERT INTO \"%s\"\n"
+        + "SELECT TIME_PARSE(a) AS __time, b AS item, c AS val\n"
+        + "FROM TABLE(\n"
+        + "  EXTERN(\n"
+        + "    '{\"type\":\"inline\",\"data\":\"2022-12-26T05:30:10,cherry,3"
+        + "\\n2022-12-26T01:45:20,banana,2\\n2022-12-26T09:10:30,apple,1\"}',\n"
+        + "    '{\"type\":\"csv\",\"findColumnsFromHeader\":false,\"columns\":[\"a\",\"b\",\"c\"]}'\n"
+        + "  )\n"
+        + ") EXTEND (a VARCHAR, b VARCHAR, c BIGINT)\n",
+        tableName
+    );
+
+    cluster.callApi().verifySqlQuery(
+        "SELECT * FROM %s",
+        tableName,
+        "apple,2022-12-26T09:00:00.000Z,1\n"
+        + "banana,2022-12-26T01:00:00.000Z,2\n"
+        + "cherry,2022-12-26T05:00:00.000Z,3"
+    );
+  }
+
   /**
    * The same layout without SEALED: a column the query produces but the table does not declare is stored after the
    * declared layout rather than rejected or dropped, and the clustering the table declared is unchanged.

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IndexTaskTest.java

@@ -19,9 +19,15 @@
 
 package org.apache.druid.testing.embedded.indexing;
 
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TableProjectionSpec;
+import org.apache.druid.indexer.granularity.SegmentGranularitySpec;
 import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskBuilder;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.java.util.common.guava.Comparators;
 import org.apache.druid.testing.embedded.EmbeddedBroker;
 import org.apache.druid.testing.embedded.EmbeddedClusterApis;
@@ -106,6 +112,56 @@ public void test_runIndexTask_forInlineDatasource()
     Assertions.assertEquals("10", cluster.runSql("SELECT COUNT(*) FROM %s", dataSource));
   }
 
+  /**
+   * Batch ingestion driven by a plain {@link TableProjectionSpec} base table instead of the legacy top-level schema
+   * fields. The spec lowers into the standard ingest inputs, so declared column order becomes segment sort order
+   * (item before __time here — not the legacy time-first sort) and the spec's query-granularity carrier floors
+   * {@code __time}, both observable in the ingested segment.
+   */
+  @Test
+  @Timeout(60)
+  public void test_runIndexTask_withBaseTableSpec()
+  {
+    final String taskId = EmbeddedClusterApis.newTaskId(dataSource);
+    final TableProjectionSpec baseTable = TableProjectionSpec
+        .builder()
+        .columns(
+            new StringDimensionSchema("item"),
+            new LongDimensionSchema("__time"),
+            new LongDimensionSchema("value")
+        )
+        .build()
+        .withQueryGranularity(Granularities.HOUR);
+    final IndexTask task = TaskBuilder
+        .ofTypeIndex()
+        .isoTimestampColumn("time")
+        .csvInputFormatWithColumns("time", "item", "value")
+        .inlineInputSourceWithData(
+            "2025-06-01T05:30:00.000Z,cherry,3"
+            + "\n2025-06-01T01:45:00.000Z,banana,2"
+            + "\n2025-06-01T09:10:00.000Z,apple,1"
+        )
+        .dataSchema(
+            schema -> schema
+                .withBaseTable(baseTable)
+                .withSegmentGranularity(new SegmentGranularitySpec(Granularities.DAY, null))
+        )
+        .dataSource(dataSource)
+        .withId(taskId);
+
+    cluster.callApi().runTask(task, overlord);
+    cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker);
+
+    // A scan without ORDER BY returns rows in segment order: sorted by item first (declared order), with __time
+    // floored to the hour by the carrier, even though the rows arrived unsorted with unfloored timestamps.
+    Assertions.assertEquals(
+        "apple,1,2025-06-01T09:00:00.000Z"
+        + "\nbanana,2,2025-06-01T01:00:00.000Z"
+        + "\ncherry,3,2025-06-01T05:00:00.000Z",
+        cluster.runSql("SELECT item, \"value\", __time FROM %s", dataSource)
+    );
+  }
+
   @Test
   @Timeout(20)
   public void test_run10Tasks_concurrently()

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

@@ -30,7 +30,6 @@
 import org.apache.druid.error.InvalidInput;
 import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.java.util.common.granularity.Granularity;
-import org.apache.druid.java.util.common.granularity.PeriodGranularity;
 import org.apache.druid.query.OrderBy;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.filter.DimFilter;
@@ -41,7 +40,6 @@
 import org.apache.druid.segment.projections.AggregateProjectionSchema;
 import org.apache.druid.segment.projections.Projections;
 import org.apache.druid.utils.CollectionUtils;
-import org.joda.time.DateTimeZone;
 
 import javax.annotation.Nullable;
 import java.util.Arrays;
@@ -248,16 +246,14 @@ private static ProjectionOrdering computeOrdering(
           continue;
         }
         final VirtualColumn vc = virtualColumns.getVirtualColumn(groupingColumn.getName());
-        final Granularity maybeGranularity = Granularities.fromVirtualColumn(vc);
+        final Granularity maybeGranularity = Granularities.fromTimeVirtualColumn(vc);
         if (maybeGranularity == null || maybeGranularity.equals(Granularities.ALL)) {
           // no __time in inputs or not supported, skip
         } else if (Granularities.NONE.equals(maybeGranularity)) {
           timeColumnName = groupingColumn.getName();
           // already found exact __time grouping, skip assigning, granularity = Granularities.NONE;
           break;
-        } else if (maybeGranularity.getClass().equals(PeriodGranularity.class)
-            && maybeGranularity.getTimeZone().equals(DateTimeZone.UTC)
-            && ((PeriodGranularity) maybeGranularity).getOrigin() == null
+        } else if (Granularities.isStandardUtcPeriod(maybeGranularity)
             && (granularity == null || maybeGranularity.isFinerThan(granularity))) {
           // found a finer period granularity than the existing granularity, or it's the first one
           timeColumnName = groupingColumn.getName();

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

@@ -21,15 +21,18 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.java.util.common.granularity.Granularity;
 import org.apache.druid.query.OrderBy;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.segment.VirtualColumn;
 import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
 import org.apache.druid.segment.projections.BaseTableProjectionSchema;
 
 import javax.annotation.Nullable;
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -47,6 +50,7 @@
  */
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes({
+    @JsonSubTypes.Type(name = TableProjectionSpec.TYPE_NAME, value = TableProjectionSpec.class),
     @JsonSubTypes.Type(
         name = ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME,
         value = ClusteredValueGroupsBaseTableProjectionSpec.class
@@ -72,10 +76,7 @@ default Granularity getQueryGranularity()
   {
     final VirtualColumn granularityVirtualColumn =
         getVirtualColumns().getVirtualColumn(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME);
-    if (granularityVirtualColumn == null) {
-      return Granularities.NONE;
-    }
-    final Granularity granularity = Granularities.fromVirtualColumn(granularityVirtualColumn);
+    final Granularity granularity = Granularities.fromTimeVirtualColumn(granularityVirtualColumn);
     return granularity == null ? Granularities.NONE : granularity;
   }
 
@@ -104,4 +105,45 @@ default boolean hasEqualCompactionState(BaseTableProjectionSpec other)
   {
     return equals(other);
   }
+
+  /**
+   * Validates that {@code queryGranularity} can be used by a base-table spec: only period granularities in the UTC
+   * time zone without an origin are currently allowed.
+   */
+  static void validateQueryGranularity(Granularity queryGranularity, String typeName)
+  {
+    if (!Granularities.isStandardUtcPeriod(queryGranularity)) {
+      throw InvalidInput.exception(
+          "Query granularity[%s] is not supported for [%s] base tables; only period granularities in the UTC time"
+          + " zone without an origin are supported",
+          queryGranularity,
+          typeName
+      );
+    }
+  }
+
+  /**
+   * Validates a {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} supplied directly to a spec (catalog JSON, or
+   * a translated DDL body): it must decode to a granularity {@link #validateQueryGranularity} accepts, so a
+   * spec cannot be constructed claiming a granularity that reads back as something else.
+   */
+  static void validateGranularity(VirtualColumn granularityColumn, String typeName)
+  {
+    if (!Collections.singletonList(ColumnHolder.TIME_COLUMN_NAME).equals(granularityColumn.requiredColumns())) {
+      throw InvalidInput.exception(
+          "virtual column [%s] must be computed from [%s] alone, but reads %s",
+          Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME,
+          ColumnHolder.TIME_COLUMN_NAME,
+          granularityColumn.requiredColumns()
+      );
+    }
+    final Granularity granularity = Granularities.fromTimeVirtualColumn(granularityColumn);
+    if (granularity == null || Granularities.NONE.equals(granularity) || Granularities.ALL.equals(granularity)) {
+      throw InvalidInput.exception(
+          "virtual column [%s] does not encode a query granularity",
+          Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+      );
+    }
+    validateQueryGranularity(granularity, typeName);
+  }
 }

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

@@ -184,7 +184,8 @@ public DimensionsSpec getDimensionsSpec()
    * {@code null} or {@code NONE} granularity is a no-op (no flooring), so this returns {@code this} unchanged.
    * <p>
    * {@code ALL} is rejected: it would floor {@code __time} to a single constant (the interval start) for the whole
-   * segment, which clustered base tables do not yet support.
+   * segment, which clustered base tables do not yet support. Everything other than a period granularity in the UTC
+   * time zone is rejected too, see {@link BaseTableProjectionSpec#validateQueryGranularity}.
    * <p>
    * Idempotent: if the spec already declares a query-granularity virtual column, that one is authoritative and this is
    * a no-op. (The compaction path attaches the virtual column up front; the MSQ generation path then calls this again
@@ -203,6 +204,7 @@ public ClusteredValueGroupsBaseTableProjectionSpec withQueryGranularity(@Nullabl
         || virtualColumns.getVirtualColumn(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME) != null) {
       return this;
     }
+    BaseTableProjectionSpec.validateQueryGranularity(queryGranularity, TYPE_NAME);
     final VirtualColumn granularityVirtualColumn =
         Granularities.toVirtualColumn(queryGranularity, Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME);
     final List<VirtualColumn> merged = new ArrayList<>(Arrays.asList(virtualColumns.getVirtualColumns()));
@@ -378,6 +380,11 @@ private static void validateVirtualColumns(VirtualColumns virtualColumns, List<D
     if (all.length == 0) {
       return;
     }
+    final VirtualColumn granularity =
+        virtualColumns.getVirtualColumn(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME);
+    if (granularity != null) {
+      BaseTableProjectionSpec.validateGranularity(granularity, TYPE_NAME);
+    }
     // Declared column types, doubling as the ColumnInspector used to infer each materialized virtual column's output
     // type (an expression's output type can depend on its input column types).
     final RowSignature.Builder signatureBuilder = RowSignature.builder();