<p dir="auto">Hey,</p>
<p dir="auto">on druid <code class="notranslate">0.10.1</code> I'm having troubles cleaning up segments from unwanted dimensions using hadoop indexing task. Discussed in here <a href="https://groups.google.com/forum/#!topic/druid-user/aBbuMYNRID8" rel="nofollow">https://groups.google.com/forum/#!topic/druid-user/aBbuMYNRID8</a></p>
<p dir="auto">I tried both schemaless approach <code class="notranslate">dimensionExclusions</code>  <a href="https://pastebin.com/raw/DwXw8dmE" rel="nofollow">https://pastebin.com/raw/DwXw8dmE</a> and <code class="notranslate">dimensions</code> <a href="https://pastebin.com/raw/cxFBYn7q" rel="nofollow">https://pastebin.com/raw/cxFBYn7q</a> ... segments get reindexed but they contain all those unwanted dimensions.</p>
<p dir="auto">At first I thought that the dimensions are not picked up from ParseSpec, so I also tried to specify it in <code class="notranslate">IoConfig.InputSpec.IngestionSpec.dimension</code> too, but that didn't help.</p>
<p dir="auto">Everyone likes doing things with SQL, so let's make it so people can do batch ingestion using SQL INSERT! I'd like to make it possible to write an <a href="https://support.microsoft.com/en-us/office/insert-into-statement-e12fd58b-33b1-479c-8221-e9f245ab8d24" rel="nofollow">INSERT INTO .... SELECT</a> query that maps onto Druid's existing batch ingestion capabilities.</p>
<p dir="auto">An example query:</p>
<div class="highlight highlight-source-sql notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="INSERT INTO tbl
SELECT
  TIME_PARSE(&quot;timestamp&quot;) AS __time,
  channel,
  cityName,
  countryName
FROM TABLE(
    EXTERN(
      '{&quot;type&quot;: &quot;s3&quot;, &quot;uris&quot;: [&quot;s3://bucket/file&quot;]}',
      '{&quot;type&quot;: &quot;json&quot;}',
      '[{&quot;name&quot;: &quot;channel&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;cityName&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;countryName&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;timestamp&quot;, &quot;type&quot;: &quot;string&quot;}]'
    )
  )
PARTITION BY FLOOR(__time TO DAY) -- was BUCKET BY in an earlier version of the proposal
CLUSTER BY channel -- was ORDER BY in an earlier version of the proposal"><pre class="notranslate"><span class="pl-k">INSERT INTO</span> tbl
<span class="pl-k">SELECT</span>
  TIME_PARSE(<span class="pl-s"><span class="pl-pds">"</span>timestamp<span class="pl-pds">"</span></span>) <span class="pl-k">AS</span> __time,
  channel,
  cityName,
  countryName
<span class="pl-k">FROM</span> TABLE(
    EXTERN(
      <span class="pl-s"><span class="pl-pds">'</span>{"type": "s3", "uris": ["s3://bucket/file"]}<span class="pl-pds">'</span></span>,
      <span class="pl-s"><span class="pl-pds">'</span>{"type": "json"}<span class="pl-pds">'</span></span>,
      <span class="pl-s"><span class="pl-pds">'</span>[{"name": "channel", "type": "string"}, {"name": "cityName", "type": "string"}, {"name": "countryName", "type": "string"}, {"name": "timestamp", "type": "string"}]<span class="pl-pds">'</span></span>
    )
  )
PARTITION BY FLOOR(__time TO DAY) <span class="pl-c"><span class="pl-c">--</span> was BUCKET BY in an earlier version of the proposal</span>
CLUSTER BY channel <span class="pl-c"><span class="pl-c">--</span> was ORDER BY in an earlier version of the proposal</span></pre></div>
<p dir="auto">Since this work may take some time to execute, there will need to be some sort of asynchronous results API. I'm thinking a good choice would be to return an ingestion task ID immediately, so the standard Druid task APIs can be used to check its status. So the response would look like this (with object result format):</p>
<div class="highlight highlight-source-json notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="[{&quot;taskId&quot;: &quot;xyzzy&quot;}]"><pre class="notranslate">[{<span class="pl-ent">"taskId"</span>: <span class="pl-s"><span class="pl-pds">"</span>xyzzy<span class="pl-pds">"</span></span>}]</pre></div>
<p dir="auto">Some thoughts about pieces of that query.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="INSERT INTO tbl"><pre class="notranslate"><code class="notranslate">INSERT INTO tbl
</code></pre></div>
<p dir="auto">In Druid there is not really a "create datasource" concept, or a datasource-wide schema. Datasources exist when they have data, and their schema is whatever data happened to get loaded. Creating a new datasource and loading more data into an existing one are the same API. So I suggest we carry those semantics over to SQL, and do ingestions (both new-table and existing-table) with the "INSERT" command.</p>
<p dir="auto">It's possible that at some point we'll want to introduce a datasource-wide schema (or partial schema), or add the ability to create empty datasources. At that point it would make sense to also add a "CREATE TABLE" command to SQL. But I suggest we start with a versatile "INSERT".</p>
<hr>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="SELECT
  TIME_PARSE(&quot;timestamp&quot;) AS __time,
  channel,
  cityName,
  countryName"><pre class="notranslate"><code class="notranslate">SELECT
  TIME_PARSE("timestamp") AS __time,
  channel,
  cityName,
  countryName
</code></pre></div>
<p dir="auto">The SELECT column list would become the columns that get ingested.</p>
<hr>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="FROM TABLE(
    EXTERN(
      '{&quot;type&quot;: &quot;s3&quot;, &quot;uris&quot;: [&quot;s3://bucket/file&quot;]}',
      '{&quot;type&quot;: &quot;json&quot;}',
      '[{&quot;name&quot;: &quot;channel&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;cityName&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;countryName&quot;, &quot;type&quot;: &quot;string&quot;}, {&quot;name&quot;: &quot;timestamp&quot;, &quot;type&quot;: &quot;string&quot;}]'
    )
  )"><pre class="notranslate"><code class="notranslate">FROM TABLE(
    EXTERN(
      '{"type": "s3", "uris": ["s3://bucket/file"]}',
      '{"type": "json"}',
      '[{"name": "channel", "type": "string"}, {"name": "cityName", "type": "string"}, {"name": "countryName", "type": "string"}, {"name": "timestamp", "type": "string"}]'
    )
  )
</code></pre></div>
<p dir="auto">We need some way to reference external data. I suggest we start with a table function that accepts an <a href="https://druid.apache.org/docs/latest/ingestion/native-batch.html#input-sources" rel="nofollow">input source</a> and <a href="https://druid.apache.org/docs/latest/ingestion/data-formats.html#input-format" rel="nofollow">input format</a>. This example uses an S3 input source and JSON input format.</p>
<p dir="auto">The "EXTERN" function in this example also accepts a row signature. That's because the SQL planner will need column name and type information in order to validate and plan a query. I think this is OK at first, but at some point I'd like to make it possible to discover this stuff at runtime.</p>
<p dir="auto">At some point it'd be nice to have the syntax here be more SQL-y (instead of having this embedded JSON). I think it'd be possible to do that by adding a bunch of new table functions alongside the existing input sources and formats. But I thought it'd be good to start with this generic one.</p>
<hr>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="PARTITION BY FLOOR(__time TO DAY)"><pre class="notranslate"><code class="notranslate">PARTITION BY FLOOR(__time TO DAY)
</code></pre></div>
<p dir="auto">(Was BUCKET BY in an earlier version of the proposal.)</p>
<p dir="auto">We need some way to specify segment granularity. This concept splits the dataset into subsets, where each subset has a single time bucket. It's common for this concept to be called "PARTITION BY".</p>
<hr>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="CLUSTER BY channel"><pre class="notranslate"><code class="notranslate">CLUSTER BY channel
</code></pre></div>
<p dir="auto">(Was ORDER BY in an earlier version of the proposal.)</p>
<p dir="auto">We need some way to specify how segments are partitioned, and how rows are ordered within segments. CLUSTER BY seems to be a de-facto-standard way to declare how you want to colocate data with same or similar values of a key.</p>
<p dir="auto">In my experience, it's a good idea to partition and order-within-partitions using the same key, so I think it's OK to have both controlled by CLUSTER BY. But if we needed to support them using different keys, I could imagine introducing an ORDER BY in addition to CLUSTER BY.</p>
<hr>
<h3 dir="auto">Proposed changes</h3>
<p dir="auto">Specific proposed changes:</p>
<ol dir="auto">
<li>
<p dir="auto">Add parser and validator support for INSERT, including ability to authorize using WRITE DATASOURCE permissions.</p>
</li>
<li>
<p dir="auto">Add an EXTERN table function and an "external" DataSource type that represents external data. The "external" DataSource would be used by the SQL layer to represent ingestion sources, and would be used to help generate ingestion tasks, but it would not understood by the native query execution system.</p>
</li>
<li>
<p dir="auto">Structure planning such that only Scan and GroupBy are used as the native query types for INSERT. (Scan represents ingestion without rollup, GroupBy represents ingestion with rollup.)</p>
</li>
<li>
<p dir="auto">Add an "orderBy" parameter to the Scan query to encapsulate the "ORDER BY" SQL clause.</p>
</li>
<li>
<p dir="auto">Split QueryMaker into an interface so there can be one implementation that executes SELECT queries and one implementation that executes INSERT queries.</p>
</li>
<li>
<p dir="auto">Add an INSERT-oriented QueryMaker that runs Scan and GroupBy queries as batch ingestion tasks. Virtual columns are like <a href="https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html#transformspec" rel="nofollow">transformSpec</a>, aggregation functions are like <a href="https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html#metricsspec" rel="nofollow">metricsSpec</a>, GROUP BY is like <a href="https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html#dimensionsspec" rel="nofollow">dimensionsSpec</a> with rollup, BUCKET BY is like <a href="https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html#granularityspec" rel="nofollow">segmentGranularity</a>, etc.</p>
</li>
</ol>
<p dir="auto">PRs.</p>
<ul dir="auto">
<li>4: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1055284963" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/11930" data-hovercard-type="pull_request" data-hovercard-url="/apache/druid/pull/11930/hovercard" href="https://github.com/apache/druid/pull/11930">#11930</a></li>
<li>1–3, 5: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1058943788" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/11959" data-hovercard-type="pull_request" data-hovercard-url="/apache/druid/pull/11959/hovercard" href="https://github.com/apache/druid/pull/11959">#11959</a></li>
</ul>
<h3 dir="auto">Whatabouts</h3>
<p dir="auto"><em>What about UPDATE, DELETE, and ALTER TABLE?</em></p>
<p dir="auto">Those would be cool too. I think they would be great as future work. UPDATE would be a good way to trigger reindexing jobs that modify actual row values, and ALTER TABLE would be a good way to trigger reindexing jobs that modify partitioning or column types. DELETE, if we're clever, could either trigger reindexing jobs or do some metadata-only thing depending on the parameters of the DELETE.</p>
<p dir="auto"><em>What about streaming?</em></p>
<p dir="auto">Calcite (our SQL parser and planning engine) has a bunch of extensions that support streaming SQL: <a href="https://calcite.apache.org/docs/stream.html" rel="nofollow">https://calcite.apache.org/docs/stream.html</a>. I haven't studied these yet, but we may be able to use this to extend SQL to support manipulation of streaming supervisors.</p>
<p dir="auto"><em>What about query functionality that the ingestion layer does not support, like subqueries, joins, limits, etc?</em></p>
<p dir="auto">I am interested in the idea of running ingestion through a system that is capable of doing all the query functionality we know and love, which opens up the door to cool things like CREATE TABLE AS SELECT and materialized views. But this proposal isn't really about that; it's about adding SQL INSERT syntax for the existing ingestion capabilities. Consider it a first step.</p>