Fix for DML queries

amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java

@@ -32,12 +32,16 @@
 import io.cdap.plugin.common.db.DBPath;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.connector.AbstractDBSpecificConnector;
+import io.cdap.plugin.util.SqlIdentifierUtils;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
 
 import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
+import javax.annotation.Nullable;
 
 /**
  * Amazon Redshift Database Connector that connects to Amazon Redshift database via JDBC.
@@ -80,9 +84,15 @@ protected SchemaReader getSchemaReader(String sessionID) {
     return new RedshiftSchemaReader(sessionID);
   }
 
+  @Override
+  protected void validateTable(Connection connection, @Nullable String database, @Nullable String schema,
+                               String table) throws SQLException {
+    super.validateTable(connection, null, schema, table);
+  }
+
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("\"%s\".\"%s\"", schema, table);
+    return SqlIdentifierUtils.quoteAnsiIdentifier(schema) + "." + SqlIdentifierUtils.quoteAnsiIdentifier(table);
   }
 
   @Override

amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java

@@ -20,6 +20,12 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.Mockito;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 
 /**
  * Unit tests for {@link RedshiftConnector}
@@ -39,6 +45,8 @@ public class RedshiftConnectorUnitTest {
   public void getTableNameTest() {
     Assert.assertEquals("\"schema\".\"table\"",
                         CONNECTOR.getTableName("db", "schema", "table"));
+    Assert.assertEquals("\"sch\"\"ema\".\"ta\"\"ble\"",
+                        CONNECTOR.getTableName("db", "sch\"ema", "ta\"ble"));
   }
 
   @Test
@@ -67,4 +75,18 @@ public void getTableQueryTest() {
                                       tableName, 100, tableName),
                         CONNECTOR.getRandomQuery(tableName, 100));
   }
+
+  @Test
+  public void validateTablePassesNullCatalogTest() throws SQLException {
+    Connection connection = Mockito.mock(Connection.class);
+    DatabaseMetaData metaData = Mockito.mock(DatabaseMetaData.class);
+    ResultSet tables = Mockito.mock(ResultSet.class);
+    Mockito.when(connection.getMetaData()).thenReturn(metaData);
+    Mockito.when(metaData.getTables(null, "schema", "table", null)).thenReturn(tables);
+    Mockito.when(tables.next()).thenReturn(true);
+
+    CONNECTOR.validateTable(connection, "db", "schema", "table");
+
+    Mockito.verify(metaData).getTables(null, "schema", "table", null);
+  }
 }

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java

@@ -37,6 +37,7 @@
 import io.cdap.plugin.db.connector.AbstractDBSpecificConnector;
 import io.cdap.plugin.mysql.MysqlDBRecord;
 import io.cdap.plugin.mysql.MysqlSchemaReader;
+import io.cdap.plugin.util.SqlIdentifierUtils;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
 
@@ -82,7 +83,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
 
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("`%s`.`%s`", database, table);
+    return SqlIdentifierUtils.quoteMySqlIdentifier(database) + "." + SqlIdentifierUtils.quoteMySqlIdentifier(table);
   }
 
   @Override

cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java

@@ -0,0 +1,43 @@
+/*
+ * 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", "mysql", "", "public",
+                                     "project:region:instance", "db", null));
+
+  @Test
+  public void getTableNameQuotesAndEscapesIdentifiersTest() {
+    Assert.assertEquals("`db`.`table`",
+                        CONNECTOR.getTableName("db", "schema", "table"));
+    Assert.assertEquals("`d``b`.`ta``ble`",
+                        CONNECTOR.getTableName("d`b", "schema", "ta`ble"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void getTableNameRejectsNulTest() {
+    CONNECTOR.getTableName("db", "schema", "ta\0ble");
+  }
+}

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java

@@ -36,12 +36,16 @@
 import io.cdap.plugin.db.connector.AbstractDBSpecificConnector;
 import io.cdap.plugin.postgres.PostgresDBRecord;
 import io.cdap.plugin.postgres.PostgresSchemaReader;
+import io.cdap.plugin.util.SqlIdentifierUtils;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
 
 import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
+import javax.annotation.Nullable;
 
 /**
  * A CloudSQL PostgreSQL Database Connector that connects to CloudSQL PostgreSQL database via JDBC.
@@ -84,9 +88,15 @@ protected SchemaReader getSchemaReader(String sessionID) {
     return new PostgresSchemaReader(sessionID);
   }
 
+  @Override
+  protected void validateTable(Connection connection, @Nullable String database, @Nullable String schema,
+                               String table) throws SQLException {
+    super.validateTable(connection, null, schema, table);
+  }
+
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("\"%s\".\"%s\"", schema, table);
+    return SqlIdentifierUtils.quoteAnsiIdentifier(schema) + "." + SqlIdentifierUtils.quoteAnsiIdentifier(table);
   }
 
   @Override

cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java

@@ -0,0 +1,66 @@
+/*
+ * 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;
+import org.mockito.Mockito;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+/**
+ * Unit tests for {@link CloudSQLPostgreSQLConnector}.
+ */
+public class CloudSQLPostgreSQLConnectorUnitTest {
+
+  private static final CloudSQLPostgreSQLConnector CONNECTOR = new CloudSQLPostgreSQLConnector(
+    new CloudSQLPostgreSQLConnectorConfig("user", "password", "postgresql", "", "public",
+                                          "project:region:instance", "db", null));
+
+  @Test
+  public void getTableNameQuotesAndEscapesIdentifiersTest() {
+    Assert.assertEquals("\"schema\".\"table\"",
+                        CONNECTOR.getTableName("db", "schema", "table"));
+    Assert.assertEquals("\"sch\"\"ema\".\"ta\"\"ble\"",
+                        CONNECTOR.getTableName("db", "sch\"ema", "ta\"ble"));
+    Assert.assertEquals("\"pg_catalog\".\"pg_class\"\" LIMIT 0; UPDATE pg_catalog.pg_class SET relname = 'x'; --\"",
+                        CONNECTOR.getTableName(
+                          "db", "pg_catalog", "pg_class\" LIMIT 0; UPDATE pg_catalog.pg_class SET relname = 'x'; --"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void getTableNameRejectsNulTest() {
+    CONNECTOR.getTableName("db", "schema", "ta\0ble");
+  }
+
+  @Test
+  public void validateTablePassesNullCatalogTest() throws SQLException {
+    Connection connection = Mockito.mock(Connection.class);
+    DatabaseMetaData metaData = Mockito.mock(DatabaseMetaData.class);
+    ResultSet tables = Mockito.mock(ResultSet.class);
+    Mockito.when(connection.getMetaData()).thenReturn(metaData);
+    Mockito.when(metaData.getTables(null, "schema", "table", null)).thenReturn(tables);
+    Mockito.when(tables.next()).thenReturn(true);
+
+    CONNECTOR.validateTable(connection, "db", "schema", "table");
+
+    Mockito.verify(metaData).getTables(null, "schema", "table", null);
+  }
+}

database-commons/src/main/java/io/cdap/plugin/db/ConnectionConfigAccessor.java

@@ -38,6 +38,7 @@ public class ConnectionConfigAccessor {
   public static final String FETCH_SIZE = "io.cdap.plugin.db.fetch.size";
   public static final String OPERATION_NAME = "io.cdap.plugin.db.operation.name";
   public static final String RELATION_TABLE_KEY = "io.cdap.plugin.db.relation.table.key";
+  public static final String READ_ONLY = "io.cdap.plugin.db.connection.readonly";
 
   private static final Gson GSON = new Gson();
   private static final Type STRING_MAP_TYPE = new TypeToken<Map<String, String>>() { }.getType();
@@ -101,6 +102,18 @@ public boolean isAutoCommitEnabled() {
     return configuration.getBoolean(AUTO_COMMIT_ENABLED, false);
   }
 
+  /**
+   * Marks the connection opened by the input format as read-only. Used for data sampling, where no statement may
+   * ever modify the database.
+   */
+  public void setReadOnly(boolean readOnly) {
+    configuration.setBoolean(READ_ONLY, readOnly);
+  }
+
+  public boolean isReadOnly() {
+    return configuration.getBoolean(READ_ONLY, false);
+  }
+
   public void setFetchSize(Integer fetchSize) {
     configuration.setInt(FETCH_SIZE, fetchSize);
   }

database-commons/src/main/java/io/cdap/plugin/db/ReadOnlyConnection.java

@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * A JDBC {@link Connection} hardened for read-only use (connection browsing, spec generation and data sampling).
+ *
+ * <p>The wrapped connection is switched to {@code autoCommit=false} and {@code readOnly=true}. With auto-commit
+ * disabled, drivers such as PostgreSQL JDBC start every transaction with {@code BEGIN READ ONLY}, so the database
+ * itself rejects any DML/DDL that might be smuggled into a query. In addition:
+ * <ul>
+ *   <li>{@link #commit()} is turned into a {@link #rollback()} so nothing can ever be persisted through this
+ *   connection, even by callers that commit unconditionally (e.g. Hadoop's {@code DBRecordReader#close()}).</li>
+ *   <li>{@link #close()} rolls back any open transaction before closing. This matters for drivers that commit on
+ *   close by default (e.g. Oracle).</li>
+ * </ul>
+ *
+ * <p>Note: {@link Connection#setReadOnly(boolean)} is only a hint per the JDBC specification and is ignored by some
+ * drivers (e.g. SQL Server). It is used here as defense in depth; the primary protection against SQL injection is
+ * strict identifier quoting at query construction time.
+ */
+public class ReadOnlyConnection extends ForwardingConnection {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ReadOnlyConnection.class);
+
+  private ReadOnlyConnection(Connection delegate) {
+    super(delegate);
+  }
+
+  /**
+   * Configures the given connection as a read-only, non auto-commit connection and wraps it.
+   * If the connection cannot be configured, it is closed and the exception is propagated (fail closed).
+   *
+   * @param connection a freshly opened connection that is not yet in a transaction
+   * @return the hardened connection
+   * @throws SQLException if the connection cannot be switched to read-only / non auto-commit mode
+   */
+  public static Connection wrap(Connection connection) throws SQLException {
+    if (connection instanceof ReadOnlyConnection) {
+      return connection;
+    }
+    try {
+      // Order matters: disable auto-commit first so that the read-only transaction characteristic set below is not
+      // immediately ended by an auto-commit (Oracle) and is applied at the start of the next transaction (PostgreSQL).
+      connection.setAutoCommit(false);
+      connection.setReadOnly(true);
+    } catch (SQLException e) {
+      try {
+        connection.close();
+      } catch (SQLException closeException) {
+        e.addSuppressed(closeException);
+      }
+      throw e;
+    }
+    return new ReadOnlyConnection(connection);
+  }
+
+  /**
+   * Auto-commit must stay disabled so that read-only transactions are enforced by the database.
+   */
+  @Override
+  public void setAutoCommit(boolean autoCommit) throws SQLException {
+    if (autoCommit) {
+      throw new SQLException("Auto-commit cannot be enabled on a read-only connection.");
+    }
+    super.setAutoCommit(false);
+  }
+
+  /**
+   * The connection must stay read-only.
+   */
+  @Override
+  public void setReadOnly(boolean readOnly) throws SQLException {
+    if (!readOnly) {
+      throw new SQLException("Read-only mode cannot be disabled on a read-only connection.");
+    }
+    super.setReadOnly(true);
+  }
+
+  /**
+   * Nothing is ever persisted through a read-only connection: a commit is executed as a rollback.
+   */
+  @Override
+  public void commit() throws SQLException {
+    rollback();
+  }
+
+  @Override
+  public void close() throws SQLException {
+    try {
+      if (!getDelegate().isClosed()) {
+        getDelegate().rollback();
+      }
+    } catch (SQLException e) {
+      // Best effort: the connection is being closed anyway and nothing may have been started on it.
+      LOG.debug("Failed to rollback read-only connection before closing it.", e);
+    } finally {
+      super.close();
+    }
+  }
+}