From cf7191d2bc52f61cd9a81f7830c1c89f9f3c826e Mon Sep 17 00:00:00 2001
From: Gian Merlino <gianmerlino@gmail.com>
Date: Tue, 18 Jan 2022 09:34:23 -0800
Subject: [PATCH] Validate target dataSource for INSERT. (#12129)

---
 .../druid/sql/calcite/planner/DruidPlanner.java   | 15 +++++++++++++--
 .../druid/sql/calcite/CalciteInsertDmlTest.java   |  9 +++++++++
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java
index 924e56b53c6..b6840525858 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java
+++ b/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)
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java
index 8709a899a75..29f8df9722a 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java
+++ b/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()
   {