[Improve][Transform-V2] Remove SQL query validation on table name

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java

@@ -24,6 +24,9 @@
 import org.apache.seatunnel.transform.exception.TransformException;
 import org.apache.seatunnel.transform.sql.SQLEngine;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import net.sf.jsqlparser.JSQLParserException;
 import net.sf.jsqlparser.expression.Expression;
 import net.sf.jsqlparser.parser.CCJSqlParserUtil;
@@ -46,6 +49,7 @@
 import java.util.stream.Collectors;
 
 public class ZetaSQLEngine implements SQLEngine {
+    private static final Logger log = LoggerFactory.getLogger(ZetaSQLEngine.class);
     private String inputTableName;
     @Nullable private String catalogTableName;
     private SeaTunnelRowType inputRowType;
@@ -119,8 +123,11 @@ private void validateSQL(Statement statement) {
                 String tableName = table.getName();
                 if (!inputTableName.equalsIgnoreCase(tableName)
                         && !tableName.equalsIgnoreCase(catalogTableName)) {
-                    throw new IllegalArgumentException(
-                            String.format("Table name: %s not found", tableName));
+                    log.warn(
+                            "SQL table name {} is not equal to input table name {} or catalog table name {}",
+                            tableName,
+                            inputTableName,
+                            catalogTableName);
                 }
             } else {
                 throw new IllegalArgumentException("Unsupported sub table syntax");

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java

@@ -69,6 +69,21 @@ public void testScaleSupport() {
                         });
     }
 
+    @Test
+    public void testQueryWithAnyTable() {
+        SQLTransform sqlTransform =
+                new SQLTransform(
+                        ReadonlyConfig.fromMap(
+                                new HashMap<String, Object>() {
+                                    {
+                                        put("query", "select * from anyTableName");
+                                    }
+                                }),
+                        getCatalogTable());
+        TableSchema tableSchema = sqlTransform.transformTableSchema();
+        Assertions.assertEquals(4, tableSchema.getColumns().size());
+    }
+
     @Test
     public void testNotLoseSourceTypeAndOptions() {
         SQLTransform sqlTransform = new SQLTransform(READONLY_CONFIG, getCatalogTable());