dailai commented on code in PR #6343:
URL: https://github.com/apache/seatunnel/pull/6343#discussion_r1639064730
##########
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceSplitEnumerator.java:
##########
@@ -51,23 +51,31 @@ public class PaimonSourceSplitEnumerator
private final Predicate predicate;
+ private int[] projection;
Review Comment:
The projection is not used here
##########
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonPredicateConverter.java:
##########
@@ -51,48 +51,100 @@
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.select.AllColumns;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectBody;
+import net.sf.jsqlparser.statement.select.SelectExpressionItem;
+import net.sf.jsqlparser.statement.select.SelectItem;
import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.stream.IntStream;
public class SqlToPaimonPredicateConverter {
- public static Predicate convertSqlWhereToPaimonPredicate(RowType rowType,
String query) {
+ public static PlainSelect convertToPlainSelect(String query) {
+ if (StringUtils.isBlank(query)) {
+ return null;
+ }
+ Statement statement = null;
try {
- if (StringUtils.isBlank(query)) {
+ statement = CCJSqlParserUtil.parse(query);
+ } catch (JSQLParserException e) {
+ throw new IllegalArgumentException("Error parsing SQL.", e);
+ }
+ // Confirm that the SQL statement is a Select statement
+ if (!(statement instanceof Select)) {
+ throw new IllegalArgumentException("Only SELECT statements are
supported.");
+ }
+ Select select = (Select) statement;
+ SelectBody selectBody = select.getSelectBody();
+ if (!(selectBody instanceof PlainSelect)) {
+ throw new IllegalArgumentException("Only simple SELECT statements
are supported.");
+ }
+ PlainSelect plainSelect = (PlainSelect) selectBody;
+ if (plainSelect.getHaving() != null
+ || plainSelect.getGroupBy() != null
+ || plainSelect.getOrderByElements() != null
+ || plainSelect.getLimit() != null) {
+ throw new IllegalArgumentException(
+ "Only SELECT statements with WHERE clause are supported.
The Having, Group By, Order By, Limit clauses are currently unsupported.");
+ }
+ return plainSelect;
+ }
+
+ public static int[] convertSqlSelectToPaimonProjectionIndex(
+ String[] fieldNames, PlainSelect plainSelect) {
+ int[] projectionIndex = null;
+ List<SelectItem> selectItems = plainSelect.getSelectItems();
+
+ List<String> columnNames = new ArrayList<>();
+ for (SelectItem selectItem : selectItems) {
+ // Selcet * return null
+ if (selectItem instanceof AllColumns) {
return null;
}
- Statement statement = CCJSqlParserUtil.parse(query);
- // Confirm that the SQL statement is a Select statement
- if (!(statement instanceof Select)) {
- throw new IllegalArgumentException("Only SELECT statements are
supported.");
+ // Gets the selcet query fields
Review Comment:
Remove
##########
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonPredicateConverter.java:
##########
@@ -51,48 +51,100 @@
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.select.AllColumns;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectBody;
+import net.sf.jsqlparser.statement.select.SelectExpressionItem;
+import net.sf.jsqlparser.statement.select.SelectItem;
import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.stream.IntStream;
public class SqlToPaimonPredicateConverter {
- public static Predicate convertSqlWhereToPaimonPredicate(RowType rowType,
String query) {
+ public static PlainSelect convertToPlainSelect(String query) {
+ if (StringUtils.isBlank(query)) {
+ return null;
+ }
+ Statement statement = null;
try {
- if (StringUtils.isBlank(query)) {
+ statement = CCJSqlParserUtil.parse(query);
+ } catch (JSQLParserException e) {
+ throw new IllegalArgumentException("Error parsing SQL.", e);
+ }
+ // Confirm that the SQL statement is a Select statement
+ if (!(statement instanceof Select)) {
+ throw new IllegalArgumentException("Only SELECT statements are
supported.");
+ }
+ Select select = (Select) statement;
+ SelectBody selectBody = select.getSelectBody();
+ if (!(selectBody instanceof PlainSelect)) {
+ throw new IllegalArgumentException("Only simple SELECT statements
are supported.");
+ }
+ PlainSelect plainSelect = (PlainSelect) selectBody;
+ if (plainSelect.getHaving() != null
+ || plainSelect.getGroupBy() != null
+ || plainSelect.getOrderByElements() != null
+ || plainSelect.getLimit() != null) {
+ throw new IllegalArgumentException(
+ "Only SELECT statements with WHERE clause are supported.
The Having, Group By, Order By, Limit clauses are currently unsupported.");
+ }
+ return plainSelect;
+ }
+
+ public static int[] convertSqlSelectToPaimonProjectionIndex(
+ String[] fieldNames, PlainSelect plainSelect) {
+ int[] projectionIndex = null;
+ List<SelectItem> selectItems = plainSelect.getSelectItems();
+
+ List<String> columnNames = new ArrayList<>();
+ for (SelectItem selectItem : selectItems) {
+ // Selcet * return null
Review Comment:
remove this comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
