diff --git a/fesod-sheet/src/main/java/org/apache/fesod/sheet/analysis/csv/CsvExcelReadExecutor.java b/fesod-sheet/src/main/java/org/apache/fesod/sheet/analysis/csv/CsvExcelReadExecutor.java
index b2741a3ff..833aba53b 100644
--- a/fesod-sheet/src/main/java/org/apache/fesod/sheet/analysis/csv/CsvExcelReadExecutor.java
+++ b/fesod-sheet/src/main/java/org/apache/fesod/sheet/analysis/csv/CsvExcelReadExecutor.java
@@ -223,13 +223,28 @@ private void dealRecord(CSVRecord record, int rowIndex) {
                 csvReadContext.csvReadWorkbookHolder().globalConfiguration().getAutoTrim();
         Boolean autoStrip =
                 csvReadContext.csvReadWorkbookHolder().globalConfiguration().getAutoStrip();
+        List<Integer> includeColumnIndexes =
+                csvReadContext.readSheetHolder().getReadSheet().getColumnIndexes();
+
         while (cellIterator.hasNext()) {
             String cellString = cellIterator.next();
+            int currentColumnIndex = columnIndex++;
+            int targetColumnIndex;
+
+            if (includeColumnIndexes == null) {
+                targetColumnIndex = currentColumnIndex;
+            } else {
+                targetColumnIndex = includeColumnIndexes.indexOf(currentColumnIndex);
+                if (targetColumnIndex < 0) {
+                    continue;
+                }
+            }
+
             ReadCellData<String> readCellData = new ReadCellData<>();
             readCellData.setRowIndex(rowIndex);
-            readCellData.setColumnIndex(columnIndex);
 
-            // csv is an empty string of whether <code>,,</code> is read or <code>,"",</code>
+            readCellData.setColumnIndex(targetColumnIndex);
+
             if (StringUtils.isNotBlank(cellString)) {
                 readCellData.setType(CellDataTypeEnum.STRING);
                 if (autoStrip) {
@@ -242,7 +257,8 @@ private void dealRecord(CSVRecord record, int rowIndex) {
             } else {
                 readCellData.setType(CellDataTypeEnum.EMPTY);
             }
-            cellMap.put(columnIndex++, readCellData);
+
+            cellMap.put(targetColumnIndex, readCellData);
         }
 
         RowTypeEnum rowType = MapUtils.isEmpty(cellMap) ? RowTypeEnum.EMPTY : RowTypeEnum.DATA;
diff --git a/fesod-sheet/src/main/java/org/apache/fesod/sheet/read/builder/CsvReaderBuilder.java b/fesod-sheet/src/main/java/org/apache/fesod/sheet/read/builder/CsvReaderBuilder.java
index 36a6befe7..84b8cb8e9 100644
--- a/fesod-sheet/src/main/java/org/apache/fesod/sheet/read/builder/CsvReaderBuilder.java
+++ b/fesod-sheet/src/main/java/org/apache/fesod/sheet/read/builder/CsvReaderBuilder.java
@@ -117,6 +117,17 @@ public CsvReaderBuilder nullString(String nullString) {
         return this;
     }
 
+    /**
+     * Specific columns to read
+     *
+     * @param columnIndexes
+     * @return
+     */
+    public CsvReaderBuilder includeColumnIndexes(List<Integer> columnIndexes) {
+        readSheet.setColumnIndexes(columnIndexes);
+        return this;
+    }
+
     /**
      * Sets the escape character.
      *
diff --git a/fesod-sheet/src/test/java/org/apache/fesod/sheet/FesodSheetTest.java b/fesod-sheet/src/test/java/org/apache/fesod/sheet/FesodSheetTest.java
index b086c585d..8f267eb7f 100644
--- a/fesod-sheet/src/test/java/org/apache/fesod/sheet/FesodSheetTest.java
+++ b/fesod-sheet/src/test/java/org/apache/fesod/sheet/FesodSheetTest.java
@@ -23,11 +23,13 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import org.apache.commons.io.FileUtils;
 import org.apache.fesod.sheet.read.builder.ExcelReaderBuilder;
 import org.apache.fesod.sheet.read.builder.ExcelReaderSheetBuilder;
 import org.apache.fesod.sheet.read.listener.ReadListener;
@@ -253,6 +255,30 @@ void testReadSheet_withAllParams_shouldReturnBuilder() {
         Assertions.assertNotNull(builder);
     }
 
+    @Test
+    void testReadCsv_withColumnIndexes_shouldFilterColumns() throws Exception {
+
+        String csvContent = "ID,Name,Age,Gender\n2,Bob,25,Male";
+        File csvFile = tempDir.resolve("test_columns.csv").toFile();
+        FileUtils.writeStringToFile(csvFile, csvContent, StandardCharsets.UTF_8);
+
+        List<Integer> targetColumns = Arrays.asList(0, 2);
+
+        List<Map<Integer, String>> readResults = FesodSheet.read(csvFile)
+                .csv()
+                .includeColumnIndexes(targetColumns)
+                .doReadSync();
+
+        Assertions.assertNotNull(readResults);
+        Assertions.assertEquals(1, readResults.size());
+
+        Map<Integer, String> row1 = readResults.get(0);
+        Assertions.assertEquals(
+                2, row1.size(), "Should only contain the 1 filtered columns (excepting the head by default)");
+        Assertions.assertEquals("2", row1.get(0));
+        Assertions.assertEquals("25", row1.get(1));
+    }
+
     @Test
     void testReadSheet_withColumnIndexes_shouldConfigureAll() {