Move INSERT & REPLACE validation to the Calcite validator

extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogIngestionTest.java

@@ -0,0 +1,174 @@
+/*
+ * 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.catalog.sql;
+
+import org.apache.druid.catalog.CatalogException;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.table.TableBuilder;
+import org.apache.druid.catalog.storage.CatalogStorage;
+import org.apache.druid.catalog.storage.CatalogTests;
+import org.apache.druid.catalog.sync.CachedMetadataCatalog;
+import org.apache.druid.catalog.sync.MetadataCatalog;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.metadata.TestDerbyConnector;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.calcite.CalciteIngestionDmlTest;
+import org.apache.druid.sql.calcite.filtration.Filtration;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.util.CalciteTests;
+import org.apache.druid.sql.calcite.util.SqlTestFramework;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Test the use of catalog specs to drive MSQ ingestion.
+ */
+public class CatalogIngestionTest extends CalciteIngestionDmlTest
+{
+  @ClassRule
+  public static final TestDerbyConnector.DerbyConnectorRule DERBY_CONNECTION_RULE =
+      new TestDerbyConnector.DerbyConnectorRule();
+
+  /**
+   * Signature for the foo datasource after applying catalog metadata.
+   */
+  private static final RowSignature FOO_SIGNATURE = RowSignature.builder()
+      .add("__time", ColumnType.LONG)
+      .add("extra1", ColumnType.STRING)
+      .add("dim2", ColumnType.STRING)
+      .add("dim1", ColumnType.STRING)
+      .add("cnt", ColumnType.LONG)
+      .add("m1", ColumnType.DOUBLE)
+      .add("extra2", ColumnType.LONG)
+      .add("extra3", ColumnType.STRING)
+      .add("m2", ColumnType.DOUBLE)
+      .build();
+
+  private static CatalogStorage storage;
+
+  @Override
+  public CatalogResolver createCatalogResolver()
+  {
+    CatalogTests.DbFixture dbFixture = new CatalogTests.DbFixture(DERBY_CONNECTION_RULE);
+    storage = dbFixture.storage;
+    MetadataCatalog catalog = new CachedMetadataCatalog(
+        storage,
+        storage.schemaRegistry(),
+        storage.jsonMapper()
+    );
+    return new LiveCatalogResolver(catalog);
+  }
+
+  @Override
+  public void finalizeTestFramework(SqlTestFramework sqlTestFramework)
+  {
+    super.finalizeTestFramework(sqlTestFramework);
+    buildTargetDatasources();
+    buildFooDatasource();
+  }
+
+  private void buildTargetDatasources()
+  {
+    TableMetadata spec = TableBuilder.datasource("hourDs", "PT1H")
+        .build();
+    createTableMetadata(spec);
+  }
+
+  public void buildFooDatasource()
+  {
+    TableMetadata spec = TableBuilder.datasource("foo", "ALL")
+        .timeColumn()
+        .column("extra1", null)
+        .column("dim2", null)
+        .column("dim1", null)
+        .column("cnt", null)
+        .column("m1", Columns.DOUBLE)
+        .column("extra2", Columns.LONG)
+        .column("extra3", Columns.STRING)
+        .hiddenColumns(Arrays.asList("dim3", "unique_dim1"))
+        .sealed(true)
+        .build();
+    createTableMetadata(spec);
+  }
+
+  private void createTableMetadata(TableMetadata table)
+  {
+    try {
+      storage.tables().create(table);
+    }
+    catch (CatalogException e) {
+      fail(e.getMessage());
+    }
+  }
+
+  /**
+   * If the segment grain is given in the catalog then use this value is used.
+   */
+  @Test
+  public void testInsertHourGrain()
+  {
+    testIngestionQuery()
+        .sql("INSERT INTO hourDs\n" +
+             "SELECT * FROM foo")
+        .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+        .expectTarget("hourDs", FOO_SIGNATURE)
+        .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+        .expectQuery(
+            newScanQueryBuilder()
+                .dataSource("foo")
+                .intervals(querySegmentSpec(Filtration.eternity()))
+                .columns("__time", "cnt", "dim1", "dim2", "extra1", "extra2", "extra3", "m1", "m2")
+                .context(queryContextWithGranularity(Granularities.HOUR))
+                .build()
+        )
+        .verify();
+  }
+
+  /**
+   * If the segment grain is given in the catalog, and also by PARTITIONED BY, then
+   * the query value is used.
+   */
+  @Test
+  public void testInsertHourGrainWithDay()
+  {
+    testIngestionQuery()
+        .sql("INSERT INTO hourDs\n" +
+             "SELECT * FROM foo\n" +
+             "PARTITIONED BY day")
+        .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+        .expectTarget("hourDs", FOO_SIGNATURE)
+        .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+        .expectQuery(
+            newScanQueryBuilder()
+                .dataSource("foo")
+                .intervals(querySegmentSpec(Filtration.eternity()))
+                .columns("__time", "cnt", "dim1", "dim2", "extra1", "extra2", "extra3", "m1", "m2")
+                .context(queryContextWithGranularity(Granularities.DAY))
+                .build()
+        )
+        .verify();
+  }
+}

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java

@@ -162,6 +162,9 @@
 import org.apache.druid.sql.calcite.export.TestExportStorageConnectorProvider;
 import org.apache.druid.sql.calcite.external.ExternalDataSource;
 import org.apache.druid.sql.calcite.external.ExternalOperatorConversion;
+import org.apache.druid.sql.calcite.external.HttpOperatorConversion;
+import org.apache.druid.sql.calcite.external.InlineOperatorConversion;
+import org.apache.druid.sql.calcite.external.LocalOperatorConversion;
 import org.apache.druid.sql.calcite.planner.CalciteRulesManager;
 import org.apache.druid.sql.calcite.planner.CatalogResolver;
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
@@ -351,6 +354,9 @@ public void configure(Binder binder)
             binder.install(new NestedDataModule());
             NestedDataModule.registerHandlersAndSerde();
             SqlBindings.addOperatorConversion(binder, ExternalOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, HttpOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, InlineOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, LocalOperatorConversion.class);
           }
 
           @Override

sql/src/main/codegen/includes/common.ftl

@@ -107,14 +107,14 @@ SqlTypeNameSpec DruidType() :
 }
 
 // Parses the supported file formats for export.
-String FileFormat() :
+SqlIdentifier FileFormat() :
 {
   SqlNode format;
 }
 {
   format = SimpleIdentifier()
   {
-    return format.toString();
+    return (SqlIdentifier) format;
   }
 }
 

sql/src/main/codegen/includes/insert.ftl

@@ -35,7 +35,7 @@ SqlNode DruidSqlInsertEof() :
   final Pair<SqlNodeList, SqlNodeList> p;
   SqlGranularityLiteral partitionedBy = null;
   SqlNodeList clusteredBy = null;
-  String exportFileFormat = null;
+  SqlIdentifier exportFileFormat = null;
 }
 {
   (

sql/src/main/codegen/includes/replace.ftl

@@ -30,7 +30,7 @@ SqlNode DruidSqlReplaceEof() :
     SqlNodeList clusteredBy = null;
     final Pair<SqlNodeList, SqlNodeList> p;
     SqlNode replaceTimeQuery = null;
-    String exportFileFormat = null;
+    SqlIdentifier exportFileFormat = null;
 }
 {
     <REPLACE> { s = span(); }
@@ -90,7 +90,7 @@ SqlNode DruidSqlReplaceEof() :
     <EOF>
     {
       sqlInsert = new SqlInsert(s.end(source), SqlNodeList.EMPTY, destination, source, columnList);
-      return DruidSqlReplace.create(sqlInsert, partitionedBy, clusteredBy, replaceTimeQuery, exportFileFormat);
+      return DruidSqlReplace.create(sqlInsert, partitionedBy, clusteredBy, exportFileFormat, replaceTimeQuery);
     }
 }
 

sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlIngest.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.sql.calcite.parser;
 
+import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
@@ -43,7 +44,7 @@ public abstract class DruidSqlIngest extends SqlInsert
   @Nullable
   protected final SqlNodeList clusteredBy;
   @Nullable
-  private final String exportFileFormat;
+  private final SqlIdentifier exportFileFormat;
 
   public DruidSqlIngest(
       SqlParserPos pos,
@@ -53,7 +54,7 @@ public DruidSqlIngest(
       SqlNodeList columnList,
       @Nullable SqlGranularityLiteral partitionedBy,
       @Nullable SqlNodeList clusteredBy,
-      @Nullable String exportFileFormat
+      @Nullable SqlIdentifier exportFileFormat
   )
   {
     super(pos, keywords, targetTable, source, columnList);
@@ -76,7 +77,7 @@ public SqlNodeList getClusteredBy()
   }
 
   @Nullable
-  public String getExportFileFormat()
+  public SqlIdentifier getExportFileFormat()
   {
     return exportFileFormat;
   }
@@ -88,6 +89,7 @@ public List<SqlNode> getOperandList()
         .addAll(super.getOperandList())
         .add(partitionedBy)
         .add(clusteredBy)
+        .add(exportFileFormat)
         .build();
   }
 }

sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlInsert.java

@@ -19,12 +19,14 @@
 
 package org.apache.druid.sql.calcite.parser;
 
+import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.druid.sql.calcite.planner.DruidSqlIngestOperator;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -39,13 +41,13 @@ public class DruidSqlInsert extends DruidSqlIngest
   public static final String SQL_INSERT_SEGMENT_GRANULARITY = "sqlInsertSegmentGranularity";
 
   // This allows reusing super.unparse
-  public static final SqlOperator OPERATOR = SqlInsert.OPERATOR;
+  public static final SqlOperator OPERATOR = DruidSqlIngestOperator.INSERT_OPERATOR;
 
   public static DruidSqlInsert create(
       @Nonnull SqlInsert insertNode,
       @Nullable SqlGranularityLiteral partitionedBy,
       @Nullable SqlNodeList clusteredBy,
-      @Nullable String exportFileFormat
+      @Nullable SqlIdentifier exportFileFormat
   )
   {
     return new DruidSqlInsert(
@@ -74,7 +76,7 @@ public DruidSqlInsert(
       SqlNodeList columnList,
       @Nullable SqlGranularityLiteral partitionedBy,
       @Nullable SqlNodeList clusteredBy,
-      @Nullable String exportFileFormat
+      @Nullable SqlIdentifier exportFileFormat
   )
   {
     super(
@@ -110,7 +112,7 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec)
     writer.newlineAndIndent();
     if (getExportFileFormat() != null) {
       writer.keyword("AS");
-      writer.print(getExportFileFormat());
+      writer.print(getExportFileFormat().toString());
       writer.newlineAndIndent();
     }
     getSource().unparse(writer, 0, 0);