[Paimon]support projection for paimon source (#6343)

Author: TaoZex

docs/en/connector-v2/source/Paimon.md

@@ -104,7 +104,7 @@ source {
     warehouse = "/tmp/paimon"
     database = "full_type"
     table = "st_test"
-    query = "select * from st_test where c_boolean= 'true' and c_tinyint > 116 and c_smallint = 15987 or c_decimal='2924137191386439303744.39292213'"
+    query = "select c_boolean, c_tinyint from st_test where c_boolean= 'true' and c_tinyint > 116 and c_smallint = 15987 or c_decimal='2924137191386439303744.39292213'"
   }
 }
 ```
@@ -161,4 +161,5 @@ source {
 ### next version
 
 - Add Paimon Source Connector
+- Support projection for Paimon Source
 

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalog.java

@@ -45,6 +45,10 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 @Slf4j
 public class PaimonCatalog implements Catalog, PaimonTable {
@@ -124,6 +128,16 @@ public CatalogTable getTable(TablePath tablePath)
         }
     }
 
+    public CatalogTable getTableWithProjection(TablePath tablePath, int[] projectionIndex)
+            throws CatalogException, TableNotExistException {
+        try {
+            FileStoreTable paimonFileStoreTableTable = (FileStoreTable) getPaimonTable(tablePath);
+            return toCatalogTable(paimonFileStoreTableTable, tablePath, projectionIndex);
+        } catch (Exception e) {
+            throw new TableNotExistException(this.catalogName, tablePath);
+        }
+    }
+
     @Override
     public Table getPaimonTable(TablePath tablePath)
             throws CatalogException, TableNotExistException {
@@ -181,8 +195,26 @@ public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
 
     private CatalogTable toCatalogTable(
             FileStoreTable paimonFileStoreTableTable, TablePath tablePath) {
+        return toCatalogTable(paimonFileStoreTableTable, tablePath, null);
+    }
+
+    private CatalogTable toCatalogTable(
+            FileStoreTable paimonFileStoreTableTable, TablePath tablePath, int[] projectionIndex) {
         org.apache.paimon.schema.TableSchema schema = paimonFileStoreTableTable.schema();
         List<DataField> dataFields = schema.fields();
+        if (!Objects.isNull(projectionIndex)) {
+            Map<Integer, DataField> indexMap =
+                    IntStream.range(0, dataFields.size())
+                            .boxed()
+                            .collect(Collectors.toMap(i -> i, dataFields::get));
+
+            dataFields =
+                    java.util.Arrays.stream(projectionIndex)
+                            .distinct()
+                            .filter(indexMap::containsKey)
+                            .mapToObj(indexMap::get)
+                            .collect(Collectors.toList());
+        }
         TableSchema.Builder builder = TableSchema.builder();
         dataFields.forEach(
                 dataField -> {

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonConfig.java

@@ -86,12 +86,6 @@ public class PaimonConfig implements Serializable {
                     .noDefaultValue()
                     .withDescription("The table you intend to access");
 
-    public static final Option<List<String>> READ_COLUMNS =
-            Options.key("read_columns")
-                    .listType()
-                    .noDefaultValue()
-                    .withDescription("The read columns of the flink table store");
-
     @Deprecated
     public static final Option<String> HDFS_SITE_PATH =
             Options.key("hdfs_site_path")

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSource.java

@@ -29,12 +29,20 @@
 import org.apache.seatunnel.connectors.seatunnel.paimon.catalog.PaimonCatalog;
 import org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonSourceConfig;
 import org.apache.seatunnel.connectors.seatunnel.paimon.source.converter.SqlToPaimonPredicateConverter;
+import org.apache.seatunnel.connectors.seatunnel.paimon.utils.RowTypeConverter;
 
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.Table;
+import org.apache.paimon.types.RowType;
+
+import net.sf.jsqlparser.statement.select.PlainSelect;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
+
+import static org.apache.seatunnel.connectors.seatunnel.paimon.source.converter.SqlToPaimonPredicateConverter.convertSqlSelectToPaimonProjectionIndex;
+import static org.apache.seatunnel.connectors.seatunnel.paimon.source.converter.SqlToPaimonPredicateConverter.convertToPlainSelect;
 
 /** Paimon connector source class. */
 public class PaimonSource
@@ -52,6 +60,8 @@ public class PaimonSource
 
     private Predicate predicate;
 
+    private int[] projectionIndex;
+
     private CatalogTable catalogTable;
 
     public PaimonSource(ReadonlyConfig readonlyConfig, PaimonCatalog paimonCatalog) {
@@ -61,12 +71,22 @@ public PaimonSource(ReadonlyConfig readonlyConfig, PaimonCatalog paimonCatalog)
                 TablePath.of(paimonSourceConfig.getNamespace(), paimonSourceConfig.getTable());
         this.catalogTable = paimonCatalog.getTable(tablePath);
         this.paimonTable = paimonCatalog.getPaimonTable(tablePath);
-        this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
-        // TODO: We can use this to realize the column projection feature later
+
         String filterSql = readonlyConfig.get(PaimonSourceConfig.QUERY_SQL);
-        this.predicate =
-                SqlToPaimonPredicateConverter.convertSqlWhereToPaimonPredicate(
-                        this.paimonTable.rowType(), filterSql);
+        PlainSelect plainSelect = convertToPlainSelect(filterSql);
+        RowType paimonRowType = this.paimonTable.rowType();
+        String[] filedNames = paimonRowType.getFieldNames().toArray(new String[0]);
+        if (!Objects.isNull(plainSelect)) {
+            this.projectionIndex = convertSqlSelectToPaimonProjectionIndex(filedNames, plainSelect);
+            if (!Objects.isNull(projectionIndex)) {
+                this.catalogTable =
+                        paimonCatalog.getTableWithProjection(tablePath, projectionIndex);
+            }
+            this.predicate =
+                    SqlToPaimonPredicateConverter.convertSqlWhereToPaimonPredicate(
+                            paimonRowType, plainSelect);
+        }
+        seaTunnelRowType = RowTypeConverter.convert(paimonRowType, projectionIndex);
     }
 
     @Override
@@ -75,26 +95,27 @@ public String getPluginName() {
     }
 
     @Override
-    public Boundedness getBoundedness() {
-        return Boundedness.BOUNDED;
+    public List<CatalogTable> getProducedCatalogTables() {
+        return Collections.singletonList(catalogTable);
     }
 
     @Override
-    public List<CatalogTable> getProducedCatalogTables() {
-        return Collections.singletonList(catalogTable);
+    public Boundedness getBoundedness() {
+        return Boundedness.BOUNDED;
     }
 
     @Override
     public SourceReader<SeaTunnelRow, PaimonSourceSplit> createReader(
             SourceReader.Context readerContext) throws Exception {
-
-        return new PaimonSourceReader(readerContext, paimonTable, seaTunnelRowType, predicate);
+        return new PaimonSourceReader(
+                readerContext, paimonTable, seaTunnelRowType, predicate, projectionIndex);
     }
 
     @Override
     public SourceSplitEnumerator<PaimonSourceSplit, PaimonSourceState> createEnumerator(
             SourceSplitEnumerator.Context<PaimonSourceSplit> enumeratorContext) throws Exception {
-        return new PaimonSourceSplitEnumerator(enumeratorContext, paimonTable, predicate);
+        return new PaimonSourceSplitEnumerator(
+                enumeratorContext, paimonTable, predicate, projectionIndex);
     }
 
     @Override
@@ -103,6 +124,6 @@ public SourceSplitEnumerator<PaimonSourceSplit, PaimonSourceState> restoreEnumer
             PaimonSourceState checkpointState)
             throws Exception {
         return new PaimonSourceSplitEnumerator(
-                enumeratorContext, paimonTable, checkpointState, predicate);
+                enumeratorContext, paimonTable, checkpointState, predicate, projectionIndex);
     }
 }

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceReader.java

@@ -49,13 +49,19 @@ public class PaimonSourceReader implements SourceReader<SeaTunnelRow, PaimonSour
     private final SeaTunnelRowType seaTunnelRowType;
     private volatile boolean noMoreSplit;
     private final Predicate predicate;
+    private int[] projection;
 
     public PaimonSourceReader(
-            Context context, Table table, SeaTunnelRowType seaTunnelRowType, Predicate predicate) {
+            Context context,
+            Table table,
+            SeaTunnelRowType seaTunnelRowType,
+            Predicate predicate,
+            int[] projection) {
         this.context = context;
         this.table = table;
         this.seaTunnelRowType = seaTunnelRowType;
         this.predicate = predicate;
+        this.projection = projection;
     }
 
     @Override
@@ -76,6 +82,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 // read logic
                 try (final RecordReader<InternalRow> reader =
                         table.newReadBuilder()
+                                .withProjection(projection)
                                 .withFilter(predicate)
                                 .newRead()
                                 .executeFilter()

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;
+
     public PaimonSourceSplitEnumerator(
-            Context<PaimonSourceSplit> context, Table table, Predicate predicate) {
+            Context<PaimonSourceSplit> context,
+            Table table,
+            Predicate predicate,
+            int[] projection) {
         this.context = context;
         this.table = table;
         this.assignedSplit = new HashSet<>();
         this.predicate = predicate;
+        this.projection = projection;
     }
 
     public PaimonSourceSplitEnumerator(
             Context<PaimonSourceSplit> context,
             Table table,
             PaimonSourceState sourceState,
-            Predicate predicate) {
+            Predicate predicate,
+            int[] projection) {
         this.context = context;
         this.table = table;
         this.assignedSplit = sourceState.getAssignedSplits();
         this.predicate = predicate;
+        this.projection = projection;
     }
 
     @Override
@@ -154,9 +162,13 @@ private void assignSplit(int taskId) {
     /** Get all splits of table */
     private Set<PaimonSourceSplit> getTableSplits() {
         final Set<PaimonSourceSplit> tableSplits = new HashSet<>();
-        // TODO Support columns projection
         final List<Split> splits =
-                table.newReadBuilder().withFilter(predicate).newScan().plan().splits();
+                table.newReadBuilder()
+                        .withProjection(projection)
+                        .withFilter(predicate)
+                        .newScan()
+                        .plan()
+                        .splits();
         splits.forEach(split -> tableSplits.add(new PaimonSourceSplit(split)));
         return tableSplits;
     }

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonPredicateConverter.java

@@ -51,48 +51,97 @@
 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)) {
-                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.");
-            }
-            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.");
-            }
-            Expression whereExpression = plainSelect.getWhere();
-            if (Objects.isNull(whereExpression)) {
+            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) {
+            if (selectItem instanceof AllColumns) {
                 return null;
+            } else if (selectItem instanceof SelectExpressionItem) {
+                SelectExpressionItem selectExpressionItem = (SelectExpressionItem) selectItem;
+                String columnName = selectExpressionItem.getExpression().toString();
+                columnNames.add(columnName);
+            } else {
+                throw new IllegalArgumentException("Error encountered parsing query fields.");
             }
-            PredicateBuilder builder = new PredicateBuilder(rowType);
-            return parseExpressionToPredicate(builder, rowType, whereExpression);
-        } catch (JSQLParserException e) {
-            throw new IllegalArgumentException("Error parsing SQL WHERE clause", e);
         }
+
+        String[] columnNamesArray = columnNames.toArray(new String[0]);
+        projectionIndex =
+                IntStream.range(0, columnNamesArray.length)
+                        .map(
+                                i -> {
+                                    String fieldName = columnNamesArray[i];
+                                    int index = Arrays.asList(fieldNames).indexOf(fieldName);
+                                    if (index == -1) {
+                                        throw new IllegalArgumentException(
+                                                "column " + fieldName + " does not exist.");
+                                    }
+                                    return index;
+                                })
+                        .toArray();
+
+        return projectionIndex;
+    }
+
+    public static Predicate convertSqlWhereToPaimonPredicate(
+            RowType rowType, PlainSelect plainSelect) {
+        Expression whereExpression = plainSelect.getWhere();
+        if (Objects.isNull(whereExpression)) {
+            return null;
+        }
+        PredicateBuilder builder = new PredicateBuilder(rowType);
+        return parseExpressionToPredicate(builder, rowType, whereExpression);
     }
 
     private static Predicate parseExpressionToPredicate(