Validate target dataSource for INSERT. (#12129)

sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java

@@ -66,6 +66,7 @@ import org.apache.calcite.tools.Planner;
 import org.apache.calcite.tools.RelConversionException;
 import org.apache.calcite.tools.ValidationException;
 import org.apache.calcite.util.Pair;
+import org.apache.druid.common.utils.IdUtils;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.guava.BaseSequence;
 import org.apache.druid.java.util.common.guava.Sequence;
@@ -663,26 +664,36 @@ public class DruidPlanner implements Closeable
     }
 
     final SqlIdentifier tableIdentifier = (SqlIdentifier) insert.getTargetTable();
+    final String dataSource;
 
     if (tableIdentifier.names.isEmpty()) {
       // I don't think this can happen, but include a branch for it just in case.
       throw new ValidationException("INSERT requires target table.");
     } else if (tableIdentifier.names.size() == 1) {
       // Unqualified name.
-      return Iterables.getOnlyElement(tableIdentifier.names);
+      dataSource = Iterables.getOnlyElement(tableIdentifier.names);
     } else {
       // Qualified name.
       final String defaultSchemaName =
           Iterables.getOnlyElement(CalciteSchema.from(frameworkConfig.getDefaultSchema()).path(null));
 
       if (tableIdentifier.names.size() == 2 && defaultSchemaName.equals(tableIdentifier.names.get(0))) {
-        return tableIdentifier.names.get(1);
+        dataSource = tableIdentifier.names.get(1);
       } else {
         throw new ValidationException(
             StringUtils.format("Cannot INSERT into [%s] because it is not a Druid datasource.", tableIdentifier)
         );
       }
     }
+
+    try {
+      IdUtils.validateId("INSERT dataSource", dataSource);
+    }
+    catch (IllegalArgumentException e) {
+      throw new ValidationException(e.getMessage());
+    }
+
+    return dataSource;
   }
 
   private String buildSQLPlanningErrorMessage(Throwable exception)

sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java

@@ -181,6 +181,15 @@ public class CalciteInsertDmlTest extends BaseCalciteQueryTest
         .verify();
   }
 
+  @Test
+  public void testInsertIntoInvalidDataSourceName()
+  {
+    testInsertQuery()
+        .sql("INSERT INTO \"in/valid\" SELECT dim1, dim2 FROM foo")
+        .expectValidationError(SqlPlanningException.class, "INSERT dataSource cannot contain the '/' character.")
+        .verify();
+  }
+
   @Test
   public void testInsertUsingColumnList()
   {