diff --git a/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java b/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java
index fb8cac4a7..57a253ebb 100644
--- a/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java
+++ b/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java
@@ -82,7 +82,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("\"%s\".\"%s\"", schema, table);
+ return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
}
@Override
diff --git a/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java b/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java
index 94ea6004b..16c94a3f0 100644
--- a/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java
+++ b/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java
@@ -41,6 +41,15 @@ public void getTableNameTest() {
CONNECTOR.getTableName("db", "schema", "table"));
}
+ /**
+ * Unit test for getTableName() with identifiers containing double quotes
+ */
+ @Test
+ public void getTableNameEscapesQuotesTest() {
+ Assert.assertEquals("\"my\"\"schema\".\"t\"\"; DELETE FROM x; --\"",
+ CONNECTOR.getTableName("db", "my\"schema", "t\"; DELETE FROM x; --"));
+ }
+
@Test
public void getRandomQuery() {
Assert.assertEquals("SELECT * FROM TestData\n" +
diff --git a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties b/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties
index ec101ed02..f6ed2205e 100644
--- a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties
+++ b/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties
@@ -18,8 +18,7 @@ validationErrorMessage=COUNT ERROR found
errorLogsMessageInvalidTableName=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : io.cdap.cdap.etl.api.validation.ValidationException: Errors were encountered during validation. \
Table 'Table123' does not exist.. Please check the system logs for more details.
errorLogsMessageInvalidCredentials=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : io.cdap.cdap.api.exception.ProgramFailureException: SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user 'testUser'
-errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : java.io.IOException: You have an error in your SQL syntax; \
- check the manual that corresponds to your MySQL server version for the right syntax to use near 'table' at line 1. Please check the system logs for more details.
+errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid
errorMessageInvalidPassword=SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user '
errorMessagePrivateConnectionName=Enter the internal IP address of the Compute Engine VM cloudsql proxy is running on, to connect to a private
errorMessageWithBlankPassword=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user '
diff --git a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java b/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java
index 0751bd160..dccb8b047 100644
--- a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java
+++ b/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java
@@ -82,7 +82,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("`%s`.`%s`", database, table);
+ return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`');
}
@Override
diff --git a/cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java b/cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java
new file mode 100644
index 000000000..84284ceb4
--- /dev/null
+++ b/cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.cloudsql.mysql;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link CloudSQLMySQLConnector}
+ */
+public class CloudSQLMySQLConnectorUnitTest {
+
+ private static final CloudSQLMySQLConnector CONNECTOR = new CloudSQLMySQLConnector(
+ new CloudSQLMySQLConnectorConfig("user", "password", "cloudsql-mysql", "", "public",
+ "project:region:instance", "db", null));
+
+ /**
+ * Unit test for getTableName()
+ */
+ @Test
+ public void getTableNameTest() {
+ Assert.assertEquals("`db`.`table`",
+ CONNECTOR.getTableName("db", null, "table"));
+ }
+
+ /**
+ * Unit test for getTableName() with identifiers containing backticks
+ */
+ @Test
+ public void getTableNameEscapesBackticksTest() {
+ Assert.assertEquals("`my``db`.`t``; DELETE FROM x; --`",
+ CONNECTOR.getTableName("my`db", null, "t`; DELETE FROM x; --"));
+ }
+}
diff --git a/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java b/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java
index 348a1b94c..8f1921478 100644
--- a/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java
+++ b/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java
@@ -86,7 +86,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("\"%s\".\"%s\"", schema, table);
+ return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
}
@Override
diff --git a/cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java b/cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java
new file mode 100644
index 000000000..fa8cb8b6a
--- /dev/null
+++ b/cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.cloudsql.postgres;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link CloudSQLPostgreSQLConnector}
+ */
+public class CloudSQLPostgreSQLConnectorUnitTest {
+
+ private static final CloudSQLPostgreSQLConnector CONNECTOR = new CloudSQLPostgreSQLConnector(
+ new CloudSQLPostgreSQLConnectorConfig("user", "password", "cloudsql-postgresql", "", "public",
+ "project:region:instance", "db", null));
+
+ /**
+ * Unit test for getTableName()
+ */
+ @Test
+ public void getTableNameTest() {
+ Assert.assertEquals("\"schema\".\"table\"",
+ CONNECTOR.getTableName("db", "schema", "table"));
+ }
+
+ /**
+ * Unit test for getTableName() with identifiers containing double quotes, using the payload from the
+ * reported sample path SQL injection
+ */
+ @Test
+ public void getTableNameEscapesQuotesTest() {
+ String table = "pg_class\" LIMIT 0; UPDATE audit_vault.write_marker SET value='INJECTED_BY_SAMPLE_PATH' " +
+ "WHERE id=1; SELECT * FROM \"pg_catalog\".\"pg_class";
+ Assert.assertEquals("\"pg_catalog\".\"pg_class\"\" LIMIT 0; UPDATE audit_vault.write_marker SET " +
+ "value='INJECTED_BY_SAMPLE_PATH' WHERE id=1; SELECT * FROM \"\"pg_catalog\"\".\"\"pg_class\"",
+ CONNECTOR.getTableName("db", "pg_catalog", table));
+ Assert.assertEquals("\"my\"\"schema\".\"table\"",
+ CONNECTOR.getTableName("db", "my\"schema", "table"));
+ }
+}
diff --git a/database-commons/pom.xml b/database-commons/pom.xml
index 98bcb3803..2d92db70d 100644
--- a/database-commons/pom.xml
+++ b/database-commons/pom.xml
@@ -77,6 +77,10 @@
dev.failsafe
failsafe
+
+ com.github.jsqlparser
+ jsqlparser
+
diff --git a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java
index e0407203b..f1b0fdf6f 100644
--- a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java
+++ b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java
@@ -150,8 +150,17 @@ protected String getConnectionString(String database) {
}
protected String getTableName(String database, String schema, String table) {
- return schema == null ? String.format("\"%s\".\"%s\"", database, table)
- : String.format("\"%s\".\"%s\".\"%s\"", database, schema, table);
+ return schema == null ? quoteIdentifier(database, '"') + "." + quoteIdentifier(table, '"')
+ : quoteIdentifier(database, '"') + "." + quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
+ }
+
+ /**
+ * Wraps the given identifier in the given quote character, doubling any embedded quote characters so that a
+ * user supplied database, schema, table or column name can never break out of the quoted identifier.
+ */
+ public static String quoteIdentifier(String identifier, char quote) {
+ String quoteString = String.valueOf(quote);
+ return quoteString + identifier.replace(quoteString, quoteString + quoteString) + quoteString;
}
protected String getTableQuery(String database, String schema, String table) {
diff --git a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java b/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java
index 908d27eb9..9517bb2b9 100644
--- a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java
+++ b/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java
@@ -52,6 +52,7 @@
import io.cdap.plugin.util.DriverCleanup;
import io.cdap.plugin.util.RetryPolicyUtil;
import io.cdap.plugin.util.RetryUtils;
+import io.cdap.plugin.util.SelectQueryValidator;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
@@ -169,10 +170,16 @@ public Schema getSchema() throws SQLException {
}
private Schema loadSchemaFromDB(Connection connection, String query) throws SQLException {
+ String finalQuery = query.contains("$CONDITIONS") ? removeConditionsClause(query) : query;
+ // never send a query that is not a single read-only SELECT to the database
+ try {
+ SelectQueryValidator.validate(finalQuery, useSquareBracketQuotes());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Invalid Import Query: " + e.getMessage(), e);
+ }
try (Statement statement = RetryUtils.createStatementWithRetry((RetryPolicy) retryPolicy, connection,
getErrorDetailsProvider())) {
statement.setMaxRows(1);
- String finalQuery = query.contains("$CONDITIONS") ? removeConditionsClause(query) : query;
ResultSet resultSet = RetryUtils.executeQueryWithRetry((RetryPolicy) retryPolicy, statement,
finalQuery, getErrorDetailsProvider());
return Schema.recordOf("outputSchema", getSchemaReader().getSchemaFields(resultSet));
@@ -228,6 +235,36 @@ protected String getErrorDetailsProviderClassName() {
return null;
}
+ /**
+ * Returns whether square brackets quote identifiers in the database's SQL dialect, as in SQL Server.
+ * Override this method to allow such identifiers in the import and bounding queries.
+ */
+ protected boolean useSquareBracketQuotes() {
+ return false;
+ }
+
+ /**
+ * Validates that the import and bounding queries are single read-only SELECT statements before the pipeline runs,
+ * since they are executed with the credential of the connection.
+ */
+ private void validateQueries(FailureCollector collector) {
+ validateQuery(DBSourceConfig.IMPORT_QUERY, "Import Query", sourceConfig.getImportQuery(), collector);
+ validateQuery(DBSourceConfig.BOUNDING_QUERY, "Bounding Query", sourceConfig.getBoundingQuery(), collector);
+ }
+
+ private void validateQuery(String property, String label, @Nullable String query, FailureCollector collector) {
+ if (Strings.isNullOrEmpty(query)) {
+ return;
+ }
+ try {
+ SelectQueryValidator.validate(query, useSquareBracketQuotes());
+ } catch (IllegalArgumentException e) {
+ collector.addFailure(String.format("Invalid %s: %s", label, e.getMessage()),
+ "Provide a single SELECT statement that only reads data.")
+ .withConfigProperty(property);
+ }
+ }
+
/**
* Returns the DBErrorDetailsProvider instance.
* Override this method to provide a custom DBErrorDetailsProvider instance.
@@ -272,6 +309,7 @@ private Connection getConnection() {
public void prepareRun(BatchSourceContext context) throws Exception {
FailureCollector collector = context.getFailureCollector();
sourceConfig.validate(collector);
+ validateQueries(collector);
collector.getOrThrowException();
LOG.debug("pluginType = {}; pluginName = {}; connectionString = {}; importQuery = {}; " +
diff --git a/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java b/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java
new file mode 100644
index 000000000..9305d84a1
--- /dev/null
+++ b/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.util;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import net.sf.jsqlparser.JSQLParserException;
+import net.sf.jsqlparser.parser.CCJSqlParserUtil;
+import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.Statements;
+import net.sf.jsqlparser.statement.select.ParenthesedSelect;
+import net.sf.jsqlparser.statement.select.PlainSelect;
+import net.sf.jsqlparser.statement.select.Select;
+import net.sf.jsqlparser.statement.select.SetOperationList;
+import net.sf.jsqlparser.statement.select.WithItem;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.stream.Collectors;
+
+/**
+ * Validates that a user provided query is a single, read-only SELECT statement before it is executed with the
+ * credential of a database connection.
+ */
+public final class SelectQueryValidator {
+
+ private static final long PARSE_TIMEOUT_MS = 10_000L;
+ // Placeholder replaced with the split condition at runtime, it is not valid SQL on its own
+ private static final String CONDITIONS_PLACEHOLDER = "$CONDITIONS";
+ private static final ThreadFactory PARSER_THREAD_FACTORY = new ThreadFactoryBuilder()
+ .setDaemon(true).setNameFormat("select-query-validator-%d").build();
+
+ private SelectQueryValidator() {
+ }
+
+ /**
+ * Validates that the given query is a single read-only SELECT statement.
+ *
+ * @param query the query to validate
+ * @param squareBracketQuotes whether square brackets quote identifiers, as in SQL Server
+ * @throws IllegalArgumentException if the query cannot be parsed or is not a single read-only SELECT statement
+ */
+ public static void validate(String query, boolean squareBracketQuotes) {
+ String parsableQuery = query.replace(CONDITIONS_PLACEHOLDER, "1 = 1");
+ // Databases disagree on whether a backslash escapes a quote inside a string literal. Requiring a single SELECT
+ // under both interpretations makes sure a crafted string literal cannot hide a second statement.
+ validate(parsableQuery, squareBracketQuotes, false);
+ validate(parsableQuery, squareBracketQuotes, true);
+ }
+
+ private static void validate(String query, boolean squareBracketQuotes, boolean backslashEscapes) {
+ Statements statements;
+ // Use our own executor so that its thread is always released, the parser does not shut down its own executor
+ // when parsing fails.
+ ExecutorService executor = Executors.newSingleThreadExecutor(PARSER_THREAD_FACTORY);
+ try {
+ statements = CCJSqlParserUtil.parseStatements(query, executor, parser -> parser
+ .withSquareBracketQuotation(squareBracketQuotes)
+ .withBackslashEscapeCharacter(backslashEscapes)
+ .withTimeOut(PARSE_TIMEOUT_MS));
+ } catch (JSQLParserException e) {
+ throw new IllegalArgumentException(
+ "Query could not be parsed. Only a single SELECT statement is allowed. Error: " + firstLine(e), e);
+ } finally {
+ executor.shutdownNow();
+ }
+
+ // the parser returns null for an empty query
+ List nonEmptyStatements = statements == null ? Collections.emptyList()
+ : statements.stream().filter(Objects::nonNull).collect(Collectors.toList());
+ if (nonEmptyStatements.size() != 1) {
+ throw new IllegalArgumentException(String.format(
+ "Only a single SELECT statement is allowed, but found %d statements.", nonEmptyStatements.size()));
+ }
+ Statement statement = nonEmptyStatements.get(0);
+ if (!(statement instanceof Select)) {
+ throw new IllegalArgumentException(String.format(
+ "Only a SELECT statement is allowed, but found a %s statement.", statement.getClass().getSimpleName()));
+ }
+ validateSelect((Select) statement);
+ }
+
+ private static void validateSelect(Select select) {
+ if (select.getWithItemsList() != null) {
+ for (WithItem withItem : select.getWithItemsList()) {
+ validateSelect(withItem.getSelect());
+ }
+ }
+ if (select instanceof PlainSelect) {
+ PlainSelect plainSelect = (PlainSelect) select;
+ if ((plainSelect.getIntoTables() != null && !plainSelect.getIntoTables().isEmpty())
+ || plainSelect.getIntoTempTable() != null) {
+ throw new IllegalArgumentException("SELECT ... INTO is not allowed, the query must only read data.");
+ }
+ if (plainSelect.getForMode() != null) {
+ throw new IllegalArgumentException(String.format(
+ "SELECT ... FOR %s is not allowed, the query must only read data.", plainSelect.getForMode().getValue()));
+ }
+ } else if (select instanceof SetOperationList) {
+ for (Select operand : ((SetOperationList) select).getSelects()) {
+ validateSelect(operand);
+ }
+ } else if (select instanceof ParenthesedSelect) {
+ validateSelect(((ParenthesedSelect) select).getSelect());
+ } else {
+ throw new IllegalArgumentException(String.format(
+ "Unsupported SELECT statement of type %s.", select.getClass().getSimpleName()));
+ }
+ }
+
+ private static String firstLine(Exception e) {
+ String message = String.valueOf(e.getMessage());
+ int newLine = message.indexOf('\n');
+ return newLine < 0 ? message : message.substring(0, newLine);
+ }
+}
diff --git a/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java b/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java
new file mode 100644
index 000000000..f92c59c2e
--- /dev/null
+++ b/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.db.connector;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link AbstractDBSpecificConnector#quoteIdentifier(String, char)}
+ */
+public class QuoteIdentifierTest {
+
+ @Test
+ public void testPlainIdentifier() {
+ Assert.assertEquals("\"table\"", AbstractDBSpecificConnector.quoteIdentifier("table", '"'));
+ Assert.assertEquals("`table`", AbstractDBSpecificConnector.quoteIdentifier("table", '`'));
+ }
+
+ @Test
+ public void testEmbeddedQuotesAreDoubled() {
+ Assert.assertEquals("\"my\"\"table\"", AbstractDBSpecificConnector.quoteIdentifier("my\"table", '"'));
+ Assert.assertEquals("`my``table`", AbstractDBSpecificConnector.quoteIdentifier("my`table", '`'));
+ Assert.assertEquals("\"\"\"\"\"\"", AbstractDBSpecificConnector.quoteIdentifier("\"\"", '"'));
+ }
+
+ @Test
+ public void testOtherQuoteCharacterIsLeftAsIs() {
+ Assert.assertEquals("\"my`table\"", AbstractDBSpecificConnector.quoteIdentifier("my`table", '"'));
+ Assert.assertEquals("`my\"table`", AbstractDBSpecificConnector.quoteIdentifier("my\"table", '`'));
+ }
+
+ @Test
+ public void testInjectionPayloadStaysInsideIdentifier() {
+ String payload = "pg_class\" LIMIT 0; UPDATE t SET v='x' WHERE id=1; SELECT * FROM \"pg_class";
+ Assert.assertEquals("\"pg_class\"\" LIMIT 0; UPDATE t SET v='x' WHERE id=1; SELECT * FROM \"\"pg_class\"",
+ AbstractDBSpecificConnector.quoteIdentifier(payload, '"'));
+ }
+}
diff --git a/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java
new file mode 100644
index 000000000..6e04ede7c
--- /dev/null
+++ b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.util;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link SelectQueryValidator}
+ */
+public class SelectQueryValidatorTest {
+
+ @Test
+ public void testValidSelectQueries() {
+ String[] queries = {
+ "SELECT * FROM my_table",
+ "select id, name from my_table where id > 10 order by name",
+ "SELECT * FROM my_table WHERE $CONDITIONS",
+ "SELECT * FROM my_table WHERE id > 10 AND $CONDITIONS",
+ "SELECT MIN(id), MAX(id) FROM my_table",
+ "SELECT * FROM \"my schema\".\"my;table\"",
+ "SELECT * FROM \"pg_class\"\"; DELETE FROM t; --\"",
+ "SELECT * FROM `db`.`my;table`",
+ "SELECT a.id, b.name FROM a JOIN b ON a.id = b.id",
+ "SELECT * FROM (SELECT id FROM my_table) t",
+ "SELECT id FROM a UNION ALL SELECT id FROM b",
+ "WITH t AS (SELECT id FROM my_table) SELECT * FROM t",
+ "SELECT * FROM my_table WHERE name = 'a;b' -- trailing ; comment",
+ "SELECT * FROM my_table /* ; DELETE FROM t */",
+ "SELECT * FROM my_table;",
+ "SELECT TOP 10 * FROM dbo.my_table",
+ "SELECT * FROM my_table WHERE ROWNUM <= 10",
+ "SELECT * FROM my_table LIMIT 10",
+ };
+ for (String query : queries) {
+ try {
+ SelectQueryValidator.validate(query, false);
+ } catch (IllegalArgumentException e) {
+ Assert.fail(String.format("Query '%s' should be valid but failed with: %s", query, e.getMessage()));
+ }
+ }
+ }
+
+ @Test
+ public void testSquareBracketIdentifiers() {
+ SelectQueryValidator.validate("SELECT * FROM [dbo].[my;table] WHERE [id] > 1", true);
+ try {
+ SelectQueryValidator.validate("SELECT * FROM [dbo].[t]; DELETE FROM [dbo].[t]", true);
+ Assert.fail("Stacked statements with square bracket identifiers should have been rejected");
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testInvalidQueries() {
+ String[] queries = {
+ // stacked statements, as in the reported vulnerability
+ "SELECT * FROM \"pg_catalog\".\"pg_class\" LIMIT 0; UPDATE t SET v = 'x' WHERE id = 1; SELECT 1",
+ "SELECT * FROM my_table; DELETE FROM my_table",
+ "SELECT 1; SELECT 2",
+ // data modification and definition
+ "UPDATE my_table SET v = 1",
+ "DELETE FROM my_table",
+ "INSERT INTO my_table VALUES (1)",
+ "DROP TABLE my_table",
+ "TRUNCATE TABLE my_table",
+ "CREATE TABLE t (id INT)",
+ "ALTER TABLE my_table ADD COLUMN c INT",
+ "MERGE INTO t USING s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.v = s.v",
+ "GRANT SELECT ON my_table TO someone",
+ // session and database commands
+ "SET search_path TO other",
+ "USE other_db",
+ "ALTER SESSION SET NLS_DATE_FORMAT = 'YYYY-MM-DD'",
+ // stored procedures
+ "EXEC my_procedure",
+ "CALL my_procedure()",
+ // reads that write or lock
+ "SELECT * INTO new_table FROM my_table",
+ "SELECT * FROM my_table FOR UPDATE",
+ "WITH d AS (DELETE FROM my_table RETURNING *) SELECT * FROM d",
+ "SELECT * FROM my_table INTO OUTFILE '/tmp/out.csv'",
+ "SELECT * INTO OUTFILE '/tmp/out.csv' FROM my_table",
+ // a dollar quoted string (PostgreSQL) that hides a second statement
+ "SELECT $$ ' $$; DELETE FROM my_table; -- '",
+ // a string that hides a second statement when backslash escapes quotes (MySQL) but not otherwise
+ "SELECT 'a\\''; DELETE FROM my_table; -- '",
+ // a string that hides a second statement when backslash does not escape quotes (PostgreSQL)
+ "SELECT 'a\\'; DELETE FROM my_table; --'",
+ // empty and unparsable
+ "",
+ ";",
+ "SELECT FROM WHERE",
+ };
+ for (String query : queries) {
+ try {
+ SelectQueryValidator.validate(query, false);
+ Assert.fail(String.format("Query '%s' should have been rejected", query));
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+ }
+ }
+
+ @Test
+ public void testErrorMessages() {
+ assertError("SELECT 1; DELETE FROM t", "Only a single SELECT statement is allowed, but found 2 statements.");
+ assertError("DELETE FROM t", "Only a SELECT statement is allowed, but found a Delete statement.");
+ assertError("SELECT * INTO t2 FROM t", "SELECT ... INTO is not allowed, the query must only read data.");
+ assertError("SELECT * FROM t FOR UPDATE", "SELECT ... FOR UPDATE is not allowed, the query must only read data.");
+ }
+
+ private static void assertError(String query, String expectedMessage) {
+ try {
+ SelectQueryValidator.validate(query, false);
+ Assert.fail(String.format("Query '%s' should have been rejected", query));
+ } catch (IllegalArgumentException e) {
+ Assert.assertEquals(expectedMessage, e.getMessage());
+ }
+ }
+}
diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java
index c78ceca6f..5997cf627 100644
--- a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java
+++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java
@@ -116,15 +116,15 @@ protected SchemaReader getSchemaReader(String sessionID) {
@Override
protected String getTableName(String database, String schema, String table) {
if (database == null && schema == null) {
- return String.format("`%s`", table);
+ return quoteIdentifier(table, '`');
}
if (database == null) {
- return String.format("`%s`.`%s`", schema, table);
+ return quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`');
}
if (schema == null) {
- return String.format("`%s`.`%s`", database, table);
+ return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`');
}
- return String.format("`%s`.`%s`.`%s`", database, schema, table);
+ return quoteIdentifier(database, '`') + "." + quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`');
}
@Override
@@ -136,6 +136,7 @@ protected String getRandomQuery(String tableName, int limit) {
@Override
protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) {
+ String strataColumn = quoteIdentifier(strata, '`');
return String.format("WITH t_%s AS (\n" +
" SELECT *,\n" +
" ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" +
@@ -146,8 +147,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata,
"WHERE MOD(sqn_%s, GREATEST(1, CAST(c_%s / %d AS BIGINT))) = 1\n" +
"ORDER BY %s\n" +
"LIMIT %d",
- sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
- limit, strata, limit);
+ sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
+ limit, strataColumn, limit);
}
@Override
diff --git a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java
index a04dfb398..27b830c76 100644
--- a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java
+++ b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java
@@ -38,6 +38,21 @@ public void testGetTableName() {
CONNECTOR.getTableName(null, null, "my_table"));
}
+ @Test
+ public void testGetTableNameEscapesBackticks() {
+ Assert.assertEquals("`main`.`default`.`t``; DELETE FROM x; --`",
+ CONNECTOR.getTableName("main", "default", "t`; DELETE FROM x; --"));
+ Assert.assertEquals("`my``schema`.`my``table`",
+ CONNECTOR.getTableName(null, "my`schema", "my`table"));
+ }
+
+ @Test
+ public void testGetStratifiedQueryEscapesStrata() {
+ String query = CONNECTOR.getStratifiedQuery("`main`.`default`.`my_table`", 10, "id`; DROP TABLE t; --", "s1");
+ Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()"));
+ Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n"));
+ }
+
@Test
public void testGetRandomQuery() {
Assert.assertEquals("SELECT * FROM `main`.`default`.`my_table`\n" +
@@ -49,13 +64,13 @@ public void testGetRandomQuery() {
public void testGetStratifiedQuery() {
Assert.assertEquals("WITH t_s1 AS (\n" +
" SELECT *,\n" +
- " ROW_NUMBER() OVER (ORDER BY id, RAND()) AS sqn_s1,\n" +
+ " ROW_NUMBER() OVER (ORDER BY `id`, RAND()) AS sqn_s1,\n" +
" COUNT(*) OVER () AS c_s1\n" +
" FROM `main`.`default`.`my_table`\n" +
" )\n" +
"SELECT * FROM t_s1\n" +
"WHERE MOD(sqn_s1, GREATEST(1, CAST(c_s1 / 10 AS BIGINT))) = 1\n" +
- "ORDER BY id\n" +
+ "ORDER BY `id`\n" +
"LIMIT 10",
CONNECTOR.getStratifiedQuery("`main`.`default`.`my_table`", 10, "id", "s1"));
}
diff --git a/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java b/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java
index b35b025b7..8240ffee3 100644
--- a/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java
+++ b/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java
@@ -23,6 +23,7 @@
import io.cdap.plugin.db.ConnectionConfig;
import io.cdap.plugin.db.source.AbstractDBSource;
+import java.util.Locale;
import javax.annotation.Nullable;
/**
@@ -46,6 +47,18 @@ protected String createConnectionString() {
return databaseSourceConfig.connectionString;
}
+ @Override
+ protected boolean useSquareBracketQuotes() {
+ // SQL Server quotes identifiers with square brackets
+ String connectionString = databaseSourceConfig.connectionString;
+ if (connectionString == null) {
+ return false;
+ }
+ String lowerCaseConnectionString = connectionString.toLowerCase(Locale.ROOT);
+ return lowerCaseConnectionString.startsWith("jdbc:sqlserver:")
+ || lowerCaseConnectionString.startsWith("jdbc:jtds:sqlserver:");
+ }
+
/**
* Generic database source configuration.
*/
diff --git a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java
index bedb720a2..f11342461 100644
--- a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java
+++ b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java
@@ -20,16 +20,19 @@
import io.cdap.cdap.api.common.Bytes;
import io.cdap.cdap.api.data.format.StructuredRecord;
import io.cdap.cdap.api.dataset.table.Table;
+import io.cdap.cdap.datapipeline.SmartWorkflow;
import io.cdap.cdap.etl.api.batch.BatchSource;
import io.cdap.cdap.etl.mock.batch.MockSink;
import io.cdap.cdap.etl.proto.v2.ETLBatchConfig;
import io.cdap.cdap.etl.proto.v2.ETLPlugin;
import io.cdap.cdap.etl.proto.v2.ETLStage;
+import io.cdap.cdap.proto.ProgramRunStatus;
import io.cdap.cdap.proto.artifact.AppRequest;
import io.cdap.cdap.proto.id.ApplicationId;
import io.cdap.cdap.proto.id.NamespaceId;
import io.cdap.cdap.test.ApplicationManager;
import io.cdap.cdap.test.DataSetManager;
+import io.cdap.cdap.test.WorkflowManager;
import io.cdap.plugin.GenericDatabasePluginTestBase;
import io.cdap.plugin.common.Constants;
import io.cdap.plugin.db.ConnectionConfig;
@@ -41,7 +44,11 @@
import java.math.BigDecimal;
import java.math.MathContext;
import java.nio.ByteBuffer;
+import java.sql.Connection;
import java.sql.Date;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
import java.sql.Time;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
@@ -52,6 +59,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
/**
* Test for ETL using databases.
@@ -329,6 +337,61 @@ public void testUserNamePasswordCombinations() throws Exception {
deployApplication(appId, appRequest);
}
+ @Test
+ public void testNonSelectImportQueryFailsDeployment() throws Exception {
+ int rowCount = countRows("my_table");
+ String importQuery = "SELECT * FROM \"my_table\" WHERE $CONDITIONS; DELETE FROM \"my_table\"";
+ ETLPlugin sourceConfig = new ETLPlugin(
+ DatabaseConstants.PLUGIN_NAME,
+ BatchSource.PLUGIN_TYPE,
+ ImmutableMap.builder()
+ .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL())
+ .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
+ .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, importQuery)
+ .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, "SELECT MIN(ID),MAX(ID) from \"my_table\"")
+ .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, "ID")
+ .put(Constants.Reference.REFERENCE_NAME, "DBNonSelectQueryTest")
+ .build(),
+ null);
+ ETLBatchConfig etlConfig = getETLBatchConfig(sourceConfig, MockSink.getPlugin("nonSelectQueryOutput"));
+ assertDeploymentFailure(NamespaceId.DEFAULT.app("dbSourceNonSelectQueryTest"), etlConfig, DATAPIPELINE_ARTIFACT,
+ "Deploying DB Source with a non SELECT import query should have failed.");
+ Assert.assertEquals(rowCount, countRows("my_table"));
+ }
+
+ @Test
+ public void testNonSelectImportQueryMacroFailsRun() throws Exception {
+ int rowCount = countRows("my_table");
+ ETLPlugin sourceConfig = new ETLPlugin(
+ DatabaseConstants.PLUGIN_NAME,
+ BatchSource.PLUGIN_TYPE,
+ ImmutableMap.builder()
+ .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL())
+ .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
+ .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, "${importQuery}")
+ .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, "SELECT MIN(ID),MAX(ID) from \"my_table\"")
+ .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, "ID")
+ .put(Constants.Reference.REFERENCE_NAME, "DBNonSelectQueryMacroTest")
+ .build(),
+ null);
+ ApplicationManager appManager = deployETL(sourceConfig, MockSink.getPlugin("nonSelectQueryMacroOutput"),
+ DATAPIPELINE_ARTIFACT, "dbSourceNonSelectQueryMacroTest");
+ WorkflowManager workflowManager = appManager.getWorkflowManager(SmartWorkflow.NAME);
+ workflowManager.start(ImmutableMap.of(
+ "importQuery", "SELECT * FROM \"my_table\" WHERE $CONDITIONS; DELETE FROM \"my_table\""));
+ workflowManager.waitForRun(ProgramRunStatus.FAILED, 5, TimeUnit.MINUTES);
+ Assert.assertEquals(rowCount, countRows("my_table"));
+ }
+
+ private int countRows(String table) throws SQLException {
+ try (Connection connection = getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(String.format("SELECT COUNT(*) FROM \"%s\"", table))) {
+ resultSet.next();
+ return resultSet.getInt(1);
+ }
+ }
+
@Test
public void testNonExistentDBTable() throws Exception {
// source
diff --git a/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java b/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java
new file mode 100644
index 000000000..c03a64a85
--- /dev/null
+++ b/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.jdbc;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link DatabaseSource}
+ */
+public class DatabaseSourceTest {
+
+ @Test
+ public void testUseSquareBracketQuotes() {
+ Assert.assertTrue(createSource("jdbc:sqlserver://localhost:1433;databaseName=db").useSquareBracketQuotes());
+ Assert.assertTrue(createSource("jdbc:jtds:sqlserver://localhost:1433/db").useSquareBracketQuotes());
+ Assert.assertTrue(createSource("JDBC:SQLSERVER://localhost:1433").useSquareBracketQuotes());
+ Assert.assertFalse(createSource("jdbc:postgresql://localhost:5432/db").useSquareBracketQuotes());
+ Assert.assertFalse(createSource("jdbc:mysql://localhost:3306/db").useSquareBracketQuotes());
+ Assert.assertFalse(createSource(null).useSquareBracketQuotes());
+ }
+
+ private static DatabaseSource createSource(String connectionString) {
+ DatabaseSource.DatabaseSourceConfig config = new DatabaseSource.DatabaseSourceConfig();
+ config.connectionString = connectionString;
+ return new DatabaseSource(config);
+ }
+}
diff --git a/mssql-plugin/src/e2e-test/resources/errorMessage.properties b/mssql-plugin/src/e2e-test/resources/errorMessage.properties
index 1f328f4d6..f94faf41e 100644
--- a/mssql-plugin/src/e2e-test/resources/errorMessage.properties
+++ b/mssql-plugin/src/e2e-test/resources/errorMessage.properties
@@ -22,6 +22,6 @@ errorMessageInvalidsourcetable=Spark program 'phase-1' failed with error: Stage
errorMessage: SQL Exception occurred: [Message='Incorrect syntax near the keyword 'table'.', SQLState='S0001', ErrorCode='156'].. Please check the system logs for more details.
errorMessageInvalidCredentialSource=Spark program 'phase-1' failed with error: Stage 'SQL Server' encountered : java.lang.IllegalArgumentException: \
Plugin with id SQL Server:source.jdbc.sqlserver does not exist in program phase-1 of application
-errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'SQL Server' encountered : java.io.IOException: Could not find stored procedure
+errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query
blank.jdbcPluginName.message=Required property 'jdbcPluginName' has no value.
blank.connection.message=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: 'S0001', errorCode: '18456', errorMessage: SQL Exception occurred: [Message='Login failed for user '
diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java
index 7e86d97ce..4f3ba8963 100644
--- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java
+++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java
@@ -117,7 +117,8 @@ public StructuredRecord transform(LongWritable longWritable, SqlServerSourceDBRe
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("\"%s\".\"%s\".\"%s\"", database, schema, table);
+ return quoteIdentifier(database, '"') + "." + quoteIdentifier(schema, '"') + "."
+ + quoteIdentifier(table, '"');
}
@Override
@@ -137,6 +138,7 @@ protected String getRandomQuery(String tableName, int limit) {
@Override
protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) {
+ String strataColumn = quoteIdentifier(strata, '"');
return String.format("WITH t_%s AS (\n" +
" SELECT *,\n" +
" ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" +
@@ -146,8 +148,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata,
"SELECT TOP %d * FROM t_%s\n" +
"WHERE sqn_%s %% CAST(0.5 * ((c_%s / %d + 1) + ABS(c_%s / %d - 1)) AS bigint) = 1\n" +
"ORDER BY %s",
- sessionID, strata, sessionID, sessionID, tableName, limit, sessionID, sessionID, sessionID,
- limit, sessionID, limit, strata);
+ sessionID, strataColumn, sessionID, sessionID, tableName, limit, sessionID, sessionID,
+ sessionID, limit, sessionID, limit, strataColumn);
}
@Override
diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java
index d8d4cf39e..b6bcb9ff9 100644
--- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java
+++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java
@@ -81,6 +81,11 @@ protected String getErrorDetailsProviderClassName() {
return SqlServerErrorDetailsProvider.class.getName();
}
+ @Override
+ protected boolean useSquareBracketQuotes() {
+ return true;
+ }
+
@Override
protected DBErrorDetailsProvider getErrorDetailsProvider() {
if (dbErrorDetailsProvider == null) {
diff --git a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java
index 0163cd55e..231d072b9 100644
--- a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java
+++ b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java
@@ -34,6 +34,27 @@ public class SqlServerConnectorUnitTest {
// private static final SqlServerConnector CONNECTOR = new SqlServerConnector(new SqlServerConnectorConfig());
+ /**
+ * Unit test for getTableName() with identifiers containing double quotes
+ */
+ @Test
+ public void getTableNameEscapesQuotesTest() {
+ Assert.assertEquals("\"db\".\"dbo\".\"t\"\"; DELETE FROM x; --\"",
+ CONNECTOR.getTableName("db", "dbo", "t\"; DELETE FROM x; --"));
+ Assert.assertEquals("\"my\"\"db\".\"my\"\"schema\".\"my\"\"table\"",
+ CONNECTOR.getTableName("my\"db", "my\"schema", "my\"table"));
+ }
+
+ /**
+ * Unit test for getStratifiedQuery() with a strata column containing double quotes
+ */
+ @Test
+ public void getStratifiedQueryEscapesStrataTest() {
+ String query = CONNECTOR.getStratifiedQuery("\"db\".\"dbo\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1");
+ Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", RAND()"));
+ Assert.assertTrue(query.endsWith("ORDER BY \"id\"\"; DROP TABLE t; --\""));
+ }
+
/**
* Unit tests for getTableQuery()
*/
diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java
index 214d8650e..0f854bc88 100644
--- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java
+++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java
@@ -98,7 +98,7 @@ protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath pa
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("`%s`.`%s`", database, table);
+ return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`');
}
@Override
@@ -112,6 +112,7 @@ protected String getRandomQuery(String tableName, int limit) {
@Override
protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) {
+ String strataColumn = quoteIdentifier(strata, '`');
return String.format("WITH t_%s AS (\n" +
" SELECT *,\n" +
" ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" +
@@ -122,8 +123,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata,
"WHERE MOD(sqn_%s, GREATEST(1, CAST(c_%s / %d AS UNSIGNED))) = 1\n" +
"ORDER BY %s\n" +
"LIMIT %d",
- sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
- limit, strata, limit);
+ sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
+ limit, strataColumn, limit);
}
@Override
diff --git a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java
index df731d194..01820842e 100644
--- a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java
+++ b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java
@@ -42,6 +42,25 @@ public void getTableNameTest() {
CONNECTOR.getTableName("db", "schema", "table"));
}
+ /**
+ * Unit test for getTableName() with identifiers containing backticks
+ */
+ @Test
+ public void getTableNameEscapesBackticksTest() {
+ Assert.assertEquals("`my``db`.`t``; DELETE FROM x; --`",
+ CONNECTOR.getTableName("my`db", "schema", "t`; DELETE FROM x; --"));
+ }
+
+ /**
+ * Unit test for getStratifiedQuery() with a strata column containing backticks
+ */
+ @Test
+ public void getStratifiedQueryEscapesStrataTest() {
+ String query = CONNECTOR.getStratifiedQuery("`db`.`table`", 10, "id`; DROP TABLE t; --", "s1");
+ Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()"));
+ Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n"));
+ }
+
/**
* Unit tests for getTableQuery()
*/
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java
index 469ca45f2..a59c0688e 100644
--- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java
+++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java
@@ -135,7 +135,7 @@ protected String getConnectionString(@Nullable String database) {
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("\"%s\".\"%s\"", schema, table);
+ return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
}
@Override
@@ -157,6 +157,7 @@ protected String getRandomQuery(String tableName, int limit) {
@Override
protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) {
+ String strataColumn = quoteIdentifier(strata, '"');
return String.format("WITH t_%s AS (\n" +
" SELECT %s.*,\n" +
" ROW_NUMBER() OVER (ORDER BY %s, dbms_random.random) AS s_%s,\n" +
@@ -166,8 +167,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata,
" )\n" +
"SELECT * FROM t_%s\n" +
"WHERE MOD(s_%s, GREATEST(1, CAST(c_%s / %d AS INTEGER))) = 1 AND ROWNUM <= %d",
- sessionID, tableName, strata, sessionID, tableName, sessionID, tableName,
- strata, sessionID, sessionID, sessionID, limit, limit);
+ sessionID, tableName, strataColumn, sessionID, tableName, sessionID, tableName,
+ strataColumn, sessionID, sessionID, sessionID, limit, limit);
}
@Override
diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java
index f89a6c1d7..2441862d0 100644
--- a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java
+++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java
@@ -37,6 +37,25 @@ public void getTableNameTest() {
CONNECTOR.getTableName("db", "schema", "table"));
}
+ /**
+ * Unit test for getTableName() with identifiers containing double quotes
+ */
+ @Test
+ public void getTableNameEscapesQuotesTest() {
+ Assert.assertEquals("\"MY\"\"SCHEMA\".\"T\"\" WHERE 1=1 --\"",
+ CONNECTOR.getTableName("db", "MY\"SCHEMA", "T\" WHERE 1=1 --"));
+ }
+
+ /**
+ * Unit test for getStratifiedQuery() with a strata column containing double quotes
+ */
+ @Test
+ public void getStratifiedQueryEscapesStrataTest() {
+ String query = CONNECTOR.getStratifiedQuery("\"SCHEMA\".\"TABLE\"", 10, "ID\", 1) --", "s1");
+ Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\", dbms_random.random"));
+ Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\" )"));
+ }
+
/**
* Unit tests for getTableQuery()
*/
diff --git a/pom.xml b/pom.xml
index c739d8855..a04552767 100644
--- a/pom.xml
+++ b/pom.xml
@@ -310,6 +310,12 @@
failsafe
3.3.2
+
+
+ com.github.jsqlparser
+ jsqlparser
+ 4.9
+
org.hsqldb
hsqldb
diff --git a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties
index a6c500f1d..9dacc0b67 100644
--- a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties
+++ b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties
@@ -18,5 +18,4 @@ errorMessageInvalidSourceHost=SQL error while getting query schema: The connecti
errorMessageInvalidTableName=Table 'table' does not exist. Ensure table '"table"' is set correctly and that the
errorMessageInvalidSinkDatabase=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '3D000', errorCode: '0', errorMessage: SQL Exception occurred
errorMessageInvalidHost=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08001', errorCode: '0', errorMessage: SQL Exception occurred: [Message='The connection attempt failed.', SQLState='08001', ErrorCode='0'].'
-errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'PostgreSQL' encountered : \
- java.io.IOException: The column index is out of range: 1, number of columns: 0.. Please check the system logs for more details.
+errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid
diff --git a/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java b/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java
index 5b23ce6cd..4ad703b61 100644
--- a/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java
+++ b/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java
@@ -117,7 +117,7 @@ public StructuredRecord transform(LongWritable longWritable, PostgresDBRecord re
@Override
protected String getTableName(String database, String schema, String table) {
- return String.format("\"%s\".\"%s\"", schema, table);
+ return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
}
@Override
@@ -129,6 +129,7 @@ protected String getRandomQuery(String tableName, int limit) {
@Override
protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) {
+ String strataColumn = quoteIdentifier(strata, '"');
// this query times out on larger datasets; would benefit from optimization
return String.format("WITH t_%s AS (\n" +
" SELECT *,\n" +
@@ -140,8 +141,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata,
"WHERE mod(sqn_%s, GREATEST(1, CAST(c_%s / %d AS bigint))) = 1\n" +
"ORDER BY %s\n" +
"LIMIT %d",
- sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
- limit, strata, limit);
+ sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID,
+ limit, strataColumn, limit);
}
@Override
diff --git a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java
index f286cf574..57e6399d4 100644
--- a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java
+++ b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java
@@ -41,6 +41,28 @@ public void getTableNameTest() {
CONNECTOR.getTableName("db", "schema", "table"));
}
+ /**
+ * Unit test for getTableName() with identifiers containing double quotes
+ */
+ @Test
+ public void getTableNameEscapesQuotesTest() {
+ Assert.assertEquals("\"pg_catalog\".\"pg_class\"\" LIMIT 0; UPDATE t SET v='x'; SELECT * FROM \"\"pg_class\"",
+ CONNECTOR.getTableName("db", "pg_catalog",
+ "pg_class\" LIMIT 0; UPDATE t SET v='x'; SELECT * FROM \"pg_class"));
+ Assert.assertEquals("\"my\"\"schema\".\"my\"\"table\"",
+ CONNECTOR.getTableName("db", "my\"schema", "my\"table"));
+ }
+
+ /**
+ * Unit test for getStratifiedQuery() with a strata column containing double quotes
+ */
+ @Test
+ public void getStratifiedQueryEscapesStrataTest() {
+ String query = CONNECTOR.getStratifiedQuery("\"schema\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1");
+ Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", random()"));
+ Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\"\n"));
+ }
+
/**
* Unit tests for getTableQuery()
*/