Revert "Add SqlParser Utility to parse and validate SQL query (#1599)" (#1607)

This reverts commit 2294e24.

Author: KGaneshDatta

pom.xml

@@ -63,7 +63,6 @@
         <jtds.driver.version>1.3.1</jtds.driver.version>
         <slf4j.version>1.7.36</slf4j.version>
         <reload4j.version>1.2.19</reload4j.version>
-        <jsqlparser.version>4.9</jsqlparser.version>
         <licenses.name>Confluent Community License</licenses.name>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.package.home>target/${project.artifactId}-${project.version}-package</project.package.home>
@@ -111,11 +110,6 @@
             <groupId>com.google.re2j</groupId>
             <artifactId>re2j</artifactId>
             <version>1.7</version>
-        </dependency>
-        <dependency>
-            <groupId>com.github.jsqlparser</groupId>
-            <artifactId>jsqlparser</artifactId>
-            <version>${jsqlparser.version}</version>
         </dependency>
          <!-- JDBC drivers, only included in runtime so they get packaged -->
         <dependency>

src/main/java/io/confluent/connect/jdbc/source/TableQuerier.java

@@ -16,7 +16,6 @@
 package io.confluent.connect.jdbc.source;
 
 import io.confluent.connect.jdbc.util.LogUtil;
-import io.confluent.connect.jdbc.util.SqlParser;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -189,27 +188,6 @@ protected void recordQuery(String query) {
     }
   }
 
-  /**
-   * Returns the query string with sensitive data redacted using SQL parsing. This method preserves
-   * the query structure while masking literal values, which is useful for troubleshooting during
-   * exceptions.
-   *
-   * @return the query with sensitive data redacted, or null if query.masked is not configured or no
-   *     query is available
-   */
-  public String getRedactedQueryString() {
-    if (!shouldRedactSensitiveLogs) {
-      return null;
-    }
-    if (loggedQueryString != null) {
-      return SqlParser.redactSensitiveData(loggedQueryString);
-    }
-    if (query != null) {
-      return SqlParser.redactSensitiveData(query);
-    }
-    return null;
-  }
-
   @Override
   public int compareTo(TableQuerier other) {
     if (this.lastUpdate < other.lastUpdate) {

src/main/java/io/confluent/connect/jdbc/source/TableQuerierProcessor.java

@@ -135,14 +135,8 @@ private void handleNonTransientException(RecordDestination<SourceRecord> destina
                                            TableQuerier querier, SQLNonTransientException sqle) {
     SQLException redactedException = shouldRedactSensitiveLogs
               ? LogUtil.redactSensitiveData(sqle) : sqle;
-    log.error(
-        "Non-transient SQL exception while running query for table: {}. Query: {}"
-            + ", sqlState: {}, vendorCode: {}.",
-        querier,
-        querier.getRedactedQueryString(),
-        sqle.getSQLState(),
-        sqle.getErrorCode(),
-        redactedException);
+    log.error("Non-transient SQL exception while running query for table: {}",
+        querier, redactedException);
     resetAndRequeueHead(querier, true);
     // This task has failed, report failure to destination
     destination.failWith(new ConnectException(redactedException));
@@ -153,13 +147,8 @@ private void handleSqlException(RecordDestination<SourceRecord> destination,
     SQLException redactedException = shouldRedactSensitiveLogs
               ? LogUtil.redactSensitiveData(sqle) : sqle;
     log.error(
-        "SQL exception while running query for table: {}. Query: {},"
-            + " sqlState: {}, vendorCode: {}."
-            + " Attempting retry {} of {} attempts.",
+        "SQL exception while running query for table: {}." + " Attempting retry {} of {} attempts.",
         querier,
-        querier.getRedactedQueryString(),
-        sqle.getSQLState(),
-        sqle.getErrorCode(),
         querier.getAttemptedRetryCount() + 1,
         maxRetriesPerQuerier,
         redactedException);
@@ -174,23 +163,14 @@ private void handleSqlException(RecordDestination<SourceRecord> destination,
   }
 
   private void handleInterruptedException(TableQuerier querier, InterruptedException e) {
-    log.error(
-        "Interrupted while running query for table: {}. Query: {}",
-        querier,
-        querier.getRedactedQueryString(),
-        e);
     resetAndRequeueHead(querier, true);
     // Interruption should not be treated as a failure, just stop processing
     Thread.currentThread().interrupt();
   }
 
   private void handleThrowable(RecordDestination<SourceRecord> destination, 
                                TableQuerier querier, Throwable t) {
-    log.error(
-        "Failed to run query for table: {}. Query: {}",
-        querier,
-        querier.getRedactedQueryString(),
-        t);
+    log.error("Failed to run query for table: {}", querier, t);
     resetAndRequeueHead(querier, true);
     // This task has failed, report failure to destination
     destination.failWith(new ConnectException("Error while processing table querier", t));

src/main/java/io/confluent/connect/jdbc/util/SqlParser.java

@@ -19,8 +19,6 @@
 import io.confluent.connect.jdbc.dialect.DatabaseDialects;
 import io.confluent.connect.jdbc.source.JdbcSourceConnectorConfig;
 import io.confluent.connect.jdbc.source.JdbcSourceConnectorConfig.TransactionIsolationMode;
-import io.confluent.connect.jdbc.util.SqlParser;
-import net.sf.jsqlparser.JSQLParserException;
 import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigValue;
 import org.slf4j.Logger;
@@ -330,6 +328,7 @@ private boolean validateQueryConfigs() {
               + "table.exclude.list when using query mode"
               + " or 'query' when using table filtering mode.";
       addConfigError(JdbcSourceConnectorConfig.QUERY_CONFIG, msg);
+      addConfigError(JdbcSourceConnectorConfig.QUERY_MASKED_CONFIG, msg);
       if (!config.getTableWhitelistSet().isEmpty()) {
         addConfigError(JdbcSourceConnectorConfig.TABLE_WHITELIST_CONFIG, msg);
       }
@@ -345,8 +344,17 @@ private boolean validateQueryConfigs() {
       return false;
     }
 
-    return !config.getQuery().isPresent()
-           || validateSqlQueryStatement(config.getQuery().get());
+    if (hasQuery
+        && !validateSelectStatement(query, JdbcSourceConnectorConfig.QUERY_CONFIG)) {
+      return false;
+    }
+    if (hasQueryMasked
+        && !validateSelectStatement(
+            queryMaskedValue, JdbcSourceConnectorConfig.QUERY_MASKED_CONFIG)) {
+      return false;
+    }
+
+    return true;
   }
 
   /**
@@ -480,29 +488,18 @@ protected void addConfigError(String configName, String errorMessage) {
   }
 
   /** Validate that provided query strings start with a SELECT statement. */
-  private boolean validateSqlQueryStatement(String statement) {
+  private boolean validateSelectStatement(String statement, String configKey) {
     String trimmedStatement = statement.trim();
     if (!SELECT_STATEMENT_PATTERN.matcher(trimmedStatement).find()) {
       String msg =
-          "Only SELECT statements are supported for query config value. "
-          + "Please provide a statement that starts with SELECT.";
-      addConfigError("query", msg);
+          String.format(
+              "Only SELECT statements are supported for '%s'. Please provide "
+              + "a statement that starts with SELECT.",
+              configKey);
+      addConfigError(configKey, msg);
       log.error(msg);
       return false;
     }
-    try {
-      SqlParser.validateSqlSyntax(trimmedStatement);
-    } catch (JSQLParserException e) {
-      String msg =
-          "Invalid SQL syntax for query config value. Please provide "
-           + "a syntactically correct SELECT statement.";
-      addConfigError("query", msg);
-      log.error(
-          "SQL syntax validation failed for query config: {}",
-          msg
-      );
-      return false;
-    }
     return true;
   }
 

src/main/java/io/confluent/connect/jdbc/validation/JdbcSourceConnectorValidation.java

@@ -19,7 +19,6 @@
 import io.confluent.connect.jdbc.source.TableQuerier.QueryMode;
 import io.confluent.connect.jdbc.util.ColumnId;
 import io.confluent.connect.jdbc.util.ExpressionBuilder;
-import io.confluent.connect.jdbc.util.SqlParser;
 import io.confluent.connect.jdbc.util.TableId;
 
 import java.sql.Connection;
@@ -29,7 +28,6 @@
 import org.junit.Test;
 import org.mockito.Matchers;
 
-import static org.junit.Assert.*;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -151,38 +149,5 @@ public void testBulkTableQuerierInQueryModeWithoutSuffix() throws SQLException {
     querier.createPreparedStatement(connectionMock);
 
     verify(databaseDialectMock, times(1)).createPreparedStatement(Matchers.any(),Matchers.eq("SELECT * FROM name"));
-  }
-
-  @Test
-  public void testGetParsedQueryStringRedactsSensitiveValuesWhenRedactionDisabled() {
-    String query = "SELECT * FROM users WHERE id = 12345 AND name = 'John Doe'";
-    BulkTableQuerier querier = new BulkTableQuerier(
-        databaseDialectMock,
-        QueryMode.QUERY,
-        query,
-        null,
-        "",
-        true
-    );
-
-    String result = querier.getRedactedQueryString();
-    String expected = "SELECT * FROM users WHERE id = 0 AND name = " + SqlParser.REDACTED_STRING;
-
-    assertEquals(expected, result);
-  }
-
-  @Test
-  public void testGetParsedQueryStringReturnsNullWhenRedactionDisabled() {
-    String query = "SELECT * FROM users WHERE id = 12345 AND name = 'John Doe'";
-    BulkTableQuerier querier = new BulkTableQuerier(
-        databaseDialectMock,
-        QueryMode.QUERY,
-        query,
-        null,
-        "",
-        false
-    );
-
-    assertNull(querier.getRedactedQueryString());
-  }
+  }  
 }