diff --git a/src/main/java/net/sf/jsqlparser/parser/feature/Feature.java b/src/main/java/net/sf/jsqlparser/parser/feature/Feature.java
index 9d2131555..f13f1b07c 100644
--- a/src/main/java/net/sf/jsqlparser/parser/feature/Feature.java
+++ b/src/main/java/net/sf/jsqlparser/parser/feature/Feature.java
@@ -457,7 +457,7 @@ public enum Feature {
      */
     oracleBlock,
 
-    execute, executeExec, executeCall, executeExecute, doStatement,
+    execute, executeExec, executeCall, executeExecute, doStatement, insertBulk,
 
     /**
      * SQL "EXECUTE" statement is allowed
diff --git a/src/main/java/net/sf/jsqlparser/statement/StatementFeatureVisitor.java b/src/main/java/net/sf/jsqlparser/statement/StatementFeatureVisitor.java
index 7f13f5efc..7ee627c3e 100644
--- a/src/main/java/net/sf/jsqlparser/statement/StatementFeatureVisitor.java
+++ b/src/main/java/net/sf/jsqlparser/statement/StatementFeatureVisitor.java
@@ -61,6 +61,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.lock.LockStatement;
 import net.sf.jsqlparser.statement.merge.Merge;
 import net.sf.jsqlparser.statement.refresh.RefreshMaterializedViewStatement;
@@ -392,6 +393,13 @@ public <S> Void visit(Insert insert, S context) {
         return super.visit(insert, context);
     }
 
+    @Override
+    public <S> Void visit(InsertBulk statement, S context) {
+        analysis.claimTopLevel();
+        analysis.certain(StmtFeature.MODIFIES_DATA);
+        return super.visit(statement, context);
+    }
+
     @Override
     public <S> Void visit(Update update, S context) {
         boolean topLevel = analysis.claimTopLevel();
diff --git a/src/main/java/net/sf/jsqlparser/statement/StatementVisitor.java b/src/main/java/net/sf/jsqlparser/statement/StatementVisitor.java
index 2001533ae..efc42f5ba 100644
--- a/src/main/java/net/sf/jsqlparser/statement/StatementVisitor.java
+++ b/src/main/java/net/sf/jsqlparser/statement/StatementVisitor.java
@@ -56,6 +56,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.insert.ParenthesedInsert;
 import net.sf.jsqlparser.statement.lock.LockStatement;
 import net.sf.jsqlparser.statement.merge.Merge;
@@ -70,6 +71,14 @@
 
 public interface StatementVisitor<T> {
 
+    default <S> T visit(InsertBulk statement, S context) {
+        return null;
+    }
+
+    default void visit(InsertBulk statement) {
+        this.visit(statement, null);
+    }
+
     <S> T visit(Analyze analyze, S context);
 
     default void visit(Analyze analyze) {
diff --git a/src/main/java/net/sf/jsqlparser/statement/StatementVisitorAdapter.java b/src/main/java/net/sf/jsqlparser/statement/StatementVisitorAdapter.java
index e633adebb..d39ccb8b2 100644
--- a/src/main/java/net/sf/jsqlparser/statement/StatementVisitorAdapter.java
+++ b/src/main/java/net/sf/jsqlparser/statement/StatementVisitorAdapter.java
@@ -62,6 +62,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.insert.InsertConflictAction;
 import net.sf.jsqlparser.statement.insert.ParenthesedInsert;
 import net.sf.jsqlparser.statement.lock.LockStatement;
@@ -314,6 +315,13 @@ private <S> T visitReturningClause(ReturningClause returningClause, S context) {
         return null;
     }
 
+    @Override
+    public <S> T visit(InsertBulk statement, S context) {
+        fromItemVisitor.visitFromItem(statement.getTable(), context);
+        statement.visitExpressions(expression -> expression.accept(expressionVisitor, context));
+        return null;
+    }
+
     @Override
     public <S> T visit(ParenthesedInsert insert, S context) {
         insert.getInsert().accept(this, context);
diff --git a/src/main/java/net/sf/jsqlparser/statement/insert/InsertBulk.java b/src/main/java/net/sf/jsqlparser/statement/insert/InsertBulk.java
new file mode 100644
index 000000000..8464a4fa9
--- /dev/null
+++ b/src/main/java/net/sf/jsqlparser/statement/insert/InsertBulk.java
@@ -0,0 +1,162 @@
+/*-
+ * #%L
+ * JSQLParser library
+ * %%
+ * Copyright (C) 2004 - 2026 JSQLParser
+ * %%
+ * Dual licensed under GNU LGPL 2.1 or Apache License 2.0
+ * #L%
+ */
+package net.sf.jsqlparser.statement.insert;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.schema.Table;
+import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.StatementVisitor;
+import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
+import net.sf.jsqlparser.statement.select.OrderByElement;
+
+/** SQL Server's SQL declaration preceding a separate bulk-load data stream. */
+public class InsertBulk implements Statement {
+    private Table table;
+    private List<ColumnDefinition> columns = new ArrayList<>();
+    private List<Option> options = new ArrayList<>();
+
+    public Table getTable() {
+        return table;
+    }
+
+    public void setTable(Table table) {
+        this.table = table;
+    }
+
+    public List<ColumnDefinition> getColumns() {
+        return columns;
+    }
+
+    public void setColumns(List<ColumnDefinition> columns) {
+        this.columns = columns;
+    }
+
+    public List<Option> getOptions() {
+        return options;
+    }
+
+    public void setOptions(List<Option> options) {
+        this.options = options;
+    }
+
+    public void visitExpressions(Consumer<Expression> visitor) {
+        for (Option option : options) {
+            if (option.getValue() != null) {
+                visitor.accept(option.getValue());
+            }
+            for (OrderByElement order : option.getOrderByElements()) {
+                visitor.accept(order.getExpression());
+            }
+        }
+    }
+
+    public StringBuilder appendTo(StringBuilder builder, Consumer<Expression> expressions,
+            Consumer<OrderByElement> ordering) {
+        builder.append("INSERT BULK ").append(table).append(" (");
+        for (int i = 0; i < columns.size(); i++) {
+            if (i > 0) {
+                builder.append(", ");
+            }
+            builder.append(columns.get(i));
+        }
+        builder.append(')');
+        if (!options.isEmpty()) {
+            builder.append(" WITH (");
+            for (int i = 0; i < options.size(); i++) {
+                if (i > 0) {
+                    builder.append(", ");
+                }
+                options.get(i).appendTo(builder, expressions, ordering);
+            }
+            builder.append(')');
+        }
+        return builder;
+    }
+
+    @Override
+    public <T, S> T accept(StatementVisitor<T> visitor, S context) {
+        return visitor.visit(this, context);
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        return appendTo(builder, value -> builder.append(value), order -> builder.append(order))
+                .toString();
+    }
+
+    public static class Option {
+        public enum Kind {
+            CHECK_CONSTRAINTS, FIRE_TRIGGERS, KEEP_NULLS, TABLOCK, ALLOW_ENCRYPTED_VALUE_MODIFICATIONS, ROWS_PER_BATCH, ORDER
+        }
+
+        private Kind kind;
+        private Expression value;
+        private List<OrderByElement> orderByElements = new ArrayList<>();
+
+        public Option(Kind kind) {
+            this.kind = kind;
+        }
+
+        public Kind getKind() {
+            return kind;
+        }
+
+        public void setKind(Kind kind) {
+            this.kind = kind;
+        }
+
+        public Expression getValue() {
+            return value;
+        }
+
+        public void setValue(Expression value) {
+            this.value = value;
+        }
+
+        public List<OrderByElement> getOrderByElements() {
+            return orderByElements;
+        }
+
+        public void setOrderByElements(List<OrderByElement> orderByElements) {
+            this.orderByElements = orderByElements;
+        }
+
+        public StringBuilder appendTo(StringBuilder builder, Consumer<Expression> expressions,
+                Consumer<OrderByElement> ordering) {
+            builder.append(kind);
+            if (value != null) {
+                builder.append(" = ");
+                expressions.accept(value);
+            }
+            if (!orderByElements.isEmpty()) {
+                builder.append(" (");
+                for (int i = 0; i < orderByElements.size(); i++) {
+                    if (i > 0) {
+                        builder.append(", ");
+                    }
+                    ordering.accept(orderByElements.get(i));
+                }
+                builder.append(')');
+            }
+            return builder;
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder builder = new StringBuilder();
+            return appendTo(builder, value -> builder.append(value), order -> builder.append(order))
+                    .toString();
+        }
+    }
+}
diff --git a/src/main/java/net/sf/jsqlparser/util/TablesNamesFinder.java b/src/main/java/net/sf/jsqlparser/util/TablesNamesFinder.java
index 3c596371b..7b122470a 100644
--- a/src/main/java/net/sf/jsqlparser/util/TablesNamesFinder.java
+++ b/src/main/java/net/sf/jsqlparser/util/TablesNamesFinder.java
@@ -139,6 +139,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.insert.InsertConflictAction;
 import net.sf.jsqlparser.statement.insert.InsertDuplicateAction;
 import net.sf.jsqlparser.statement.insert.OracleMultiInsertBranch;
@@ -1571,6 +1572,13 @@ public <S> Void visit(DoStatement statement, S context) {
         return null;
     }
 
+    @Override
+    public <S> Void visit(InsertBulk statement, S context) {
+        visit(statement.getTable(), context);
+        statement.visitExpressions(value -> value.accept(this, context));
+        return null;
+    }
+
     @Override
     public void visit(CreateIndex createIndex) {
         StatementVisitor.super.visit(createIndex);
diff --git a/src/main/java/net/sf/jsqlparser/util/deparser/StatementDeParser.java b/src/main/java/net/sf/jsqlparser/util/deparser/StatementDeParser.java
index d85dbddd2..0a597211e 100644
--- a/src/main/java/net/sf/jsqlparser/util/deparser/StatementDeParser.java
+++ b/src/main/java/net/sf/jsqlparser/util/deparser/StatementDeParser.java
@@ -82,6 +82,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.insert.ParenthesedInsert;
 import net.sf.jsqlparser.statement.lock.LockStatement;
 import net.sf.jsqlparser.statement.merge.Merge;
@@ -157,6 +158,13 @@ public <S> StringBuilder visit(DoStatement statement, S context) {
         return statement.appendTo(builder, code -> code.accept(expressionDeParser, context));
     }
 
+    @Override
+    public <S> StringBuilder visit(InsertBulk statement, S context) {
+        OrderByDeParser order = new OrderByDeParser(expressionDeParser, builder);
+        return statement.appendTo(builder, value -> value.accept(expressionDeParser, context),
+                element -> order.deParseElement(element, context));
+    }
+
     @Override
     public <S> StringBuilder visit(CreateTable createTable, S context) {
         CreateTableDeParser createTableDeParser = new CreateTableDeParser(this, builder);
diff --git a/src/main/java/net/sf/jsqlparser/util/validation/feature/SqlServerVersion.java b/src/main/java/net/sf/jsqlparser/util/validation/feature/SqlServerVersion.java
index 863fbc839..75fd2bb9e 100644
--- a/src/main/java/net/sf/jsqlparser/util/validation/feature/SqlServerVersion.java
+++ b/src/main/java/net/sf/jsqlparser/util/validation/feature/SqlServerVersion.java
@@ -60,6 +60,7 @@ public enum SqlServerVersion implements Version {
 
                     // https://docs.microsoft.com/en-us/sql/t-sql/statements/insert-transact-sql?view=sql-server-ver15
                     Feature.insert,
+                    Feature.insertBulk,
                     Feature.insertValues,
                     Feature.insertFromSelect, // https://docs.microsoft.com/en-us/sql/t-sql/queries/update-transact-sql?view=sql-server-ver15
                     Feature.update,
diff --git a/src/main/java/net/sf/jsqlparser/util/validation/validator/InsertBulkValidator.java b/src/main/java/net/sf/jsqlparser/util/validation/validator/InsertBulkValidator.java
new file mode 100644
index 000000000..dcf6a9af6
--- /dev/null
+++ b/src/main/java/net/sf/jsqlparser/util/validation/validator/InsertBulkValidator.java
@@ -0,0 +1,31 @@
+/*-
+ * #%L
+ * JSQLParser library
+ * %%
+ * Copyright (C) 2004 - 2026 JSQLParser
+ * %%
+ * Dual licensed under GNU LGPL 2.1 or Apache License 2.0
+ * #L%
+ */
+package net.sf.jsqlparser.util.validation.validator;
+
+import java.util.stream.Collectors;
+import net.sf.jsqlparser.parser.feature.Feature;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
+import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
+import net.sf.jsqlparser.util.validation.ValidationCapability;
+import net.sf.jsqlparser.util.validation.metadata.NamedObject;
+
+public class InsertBulkValidator extends AbstractValidator<InsertBulk> {
+    @Override
+    public void validate(InsertBulk statement) {
+        validateFeature(Feature.insertBulk);
+        validateOptionalFromItem(statement.getTable());
+        for (ValidationCapability capability : getCapabilities()) {
+            validateOptionalColumnNames(capability, statement.getColumns().stream()
+                    .map(ColumnDefinition::getColumnName).collect(Collectors.toList()),
+                    NamedObject.table);
+        }
+        statement.visitExpressions(this::validateOptionalExpression);
+    }
+}
diff --git a/src/main/java/net/sf/jsqlparser/util/validation/validator/StatementValidator.java b/src/main/java/net/sf/jsqlparser/util/validation/validator/StatementValidator.java
index 282e30172..4bed769b1 100644
--- a/src/main/java/net/sf/jsqlparser/util/validation/validator/StatementValidator.java
+++ b/src/main/java/net/sf/jsqlparser/util/validation/validator/StatementValidator.java
@@ -83,6 +83,7 @@
 import net.sf.jsqlparser.statement.grant.Grant;
 import net.sf.jsqlparser.statement.imprt.Import;
 import net.sf.jsqlparser.statement.insert.Insert;
+import net.sf.jsqlparser.statement.insert.InsertBulk;
 import net.sf.jsqlparser.statement.insert.ParenthesedInsert;
 import net.sf.jsqlparser.statement.lock.LockStatement;
 import net.sf.jsqlparser.statement.merge.Merge;
@@ -239,6 +240,12 @@ public <S> Void visit(DoStatement statement, S context) {
         return null;
     }
 
+    @Override
+    public <S> Void visit(InsertBulk statement, S context) {
+        getValidator(InsertBulkValidator.class).validate(statement);
+        return null;
+    }
+
     @Override
     public <S> Void visit(SetStatement set, S context) {
         getValidator(SetStatementValidator.class).validate(set);
diff --git a/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt b/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt
index 1720fff54..e5a333b1d 100644
--- a/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt
+++ b/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt
@@ -2746,6 +2746,10 @@ Statement SingleStatement() :
                            && (getToken(4).kind == K_FUNCTION || getToken(4).kind == K_PROCEDURE))))) })
            stm = SqlServerRoutine()
            |
+            LOOKAHEAD({ Dialect.SQLSERVER.name().equals(getAsString(Feature.dialect))
+                    && getToken(1).kind == K_INSERT && "BULK".equalsIgnoreCase(getToken(2).image) })
+            stm = InsertBulk()
+            |
            LOOKAHEAD(3) (
                 [ LOOKAHEAD(2) with=WithList() ]
                 (
@@ -4748,6 +4752,77 @@ List<Partition> Partitions():
     }
 }
 
+InsertBulk InsertBulk():
+{
+    InsertBulk statement = new InsertBulk();
+    Table table;
+    ColumnDefinition column;
+    InsertBulk.Option option;
+    Set<InsertBulk.Option.Kind> kinds = EnumSet.noneOf(InsertBulk.Option.Kind.class);
+}
+{
+    <K_INSERT> AccessKeyword("BULK") table=Table() { statement.setTable(table); }
+    "(" column=InsertBulkColumn() { statement.getColumns().add(column); }
+    ( "," column=InsertBulkColumn() { statement.getColumns().add(column); } )* ")"
+    [ <K_WITH> "(" option=InsertBulkOption() {
+        statement.getOptions().add(option); kinds.add(option.getKind());
+      }
+      ( "," option=InsertBulkOption() {
+          requireDdlSyntax(kinds.add(option.getKind()), "Duplicate INSERT BULK option");
+          statement.getOptions().add(option);
+      } )* ")" ]
+    { return statement; }
+}
+
+ColumnDefinition InsertBulkColumn():
+{
+    String name;
+    String collation;
+    ColDataType type;
+    ColumnDefinition column;
+}
+{
+    name=RelObjectName() type=ColDataType() { column = new ColumnDefinition(name, type); }
+    [ <K_COLLATE> collation=RelObjectName() { column.setColumnSpecs(Arrays.asList("COLLATE", collation)); } ]
+    { return column; }
+}
+
+InsertBulk.Option InsertBulkOption():
+{
+    String name;
+    Token value;
+    Column column;
+    Token direction;
+    OrderByElement order;
+    InsertBulk.Option option;
+}
+{
+    (
+        <K_ORDER> { option = new InsertBulk.Option(InsertBulk.Option.Kind.ORDER); }
+        "(" column=Column() {
+            order = new OrderByElement().withExpression(column);
+            option.getOrderByElements().add(order);
+        }
+        [ (direction=<K_ASC> | direction=<K_DESC>) { order.setAsc(direction.kind == K_ASC); order.setAscDescPresent(true); } ]
+        ( "," column=Column() {
+            order = new OrderByElement().withExpression(column);
+            option.getOrderByElements().add(order);
+          }
+          [ (direction=<K_ASC> | direction=<K_DESC>) { order.setAsc(direction.kind == K_ASC); order.setAscDescPresent(true); } ]
+        )* ")"
+        |
+        LOOKAHEAD({ isKeywordAhead("ROWS_PER_BATCH") }) AccessKeyword("ROWS_PER_BATCH") "=" value=<S_LONG>
+        { option = new InsertBulk.Option(InsertBulk.Option.Kind.ROWS_PER_BATCH); option.setValue(new LongValue(value.image)); }
+        |
+        name=RelObjectName() {
+            option = new InsertBulk.Option(typeDdlEnum(InsertBulk.Option.Kind.class, name));
+            requireDdlSyntax(option.getKind() != InsertBulk.Option.Kind.ORDER
+                    && option.getKind() != InsertBulk.Option.Kind.ROWS_PER_BATCH, "Missing INSERT BULK option value");
+        }
+    )
+    { return option; }
+}
+
 Insert InsertWithWithItems( List<WithItem<?>> withItems ):
 {
     Insert insert;
diff --git a/src/site/sphinx/usage.rst b/src/site/sphinx/usage.rst
index b77dc87e4..4ab672491 100644
--- a/src/site/sphinx/usage.rst
+++ b/src/site/sphinx/usage.rst
@@ -788,6 +788,13 @@ operators such as ``js#>>'{a}'`` and ``js#>'{a}'`` work without surrounding
 spaces. Quote identifiers containing ``#``, for example ``"js#"``. Other
 dialects retain their existing identifier and hash-comment rules.
 
+``Dialect.SQLSERVER`` enables ``INSERT BULK table (name type, ...) WITH (...)``.
+``InsertBulk`` exposes the target table, existing ``ColumnDefinition`` models,
+and ordered typed options, including ``ROWS_PER_BATCH`` and ``ORDER`` keys.
+The SQL declaration is preserved; the following binary bulk-load data stream
+is outside the SQL parser. Visitors and deparsers traverse option values and
+ordering expressions. Validation uses the ``insertBulk`` capability.
+
 With ``Dialect.SQLSERVER``, ``PRIMARY KEY NONCLUSTERED (id)`` and
 ``UNIQUE CLUSTERED (id)`` store their clustering option in ``Index.getClustering()``
 for both ``CREATE TABLE`` and ``ALTER TABLE``. Without that dialect, these words
diff --git a/src/test/java/net/sf/jsqlparser/statement/insert/InsertBulkTest.java b/src/test/java/net/sf/jsqlparser/statement/insert/InsertBulkTest.java
new file mode 100644
index 000000000..bf764fd9f
--- /dev/null
+++ b/src/test/java/net/sf/jsqlparser/statement/insert/InsertBulkTest.java
@@ -0,0 +1,151 @@
+/*-
+ * #%L
+ * JSQLParser library
+ * %%
+ * Copyright (C) 2004 - 2026 JSQLParser
+ * %%
+ * Dual licensed under GNU LGPL 2.1 or Apache License 2.0
+ * #L%
+ */
+package net.sf.jsqlparser.statement.insert;
+
+import static net.sf.jsqlparser.test.TestUtils.assertSqlCanBeParsedAndDeparsed;
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import net.sf.jsqlparser.JSQLParserException;
+import net.sf.jsqlparser.expression.ExpressionVisitorAdapter;
+import net.sf.jsqlparser.expression.LongValue;
+import net.sf.jsqlparser.parser.AbstractJSqlParser.Dialect;
+import net.sf.jsqlparser.parser.CCJSqlParserUtil;
+import net.sf.jsqlparser.parser.feature.Feature;
+import net.sf.jsqlparser.parser.feature.FeatureConfiguration;
+import net.sf.jsqlparser.statement.StatementVisitorAdapter;
+import net.sf.jsqlparser.statement.select.SelectVisitorAdapter;
+import net.sf.jsqlparser.util.TablesNamesFinder;
+import net.sf.jsqlparser.util.deparser.ExpressionDeParser;
+import net.sf.jsqlparser.util.deparser.SelectDeParser;
+import net.sf.jsqlparser.util.deparser.StatementDeParser;
+import net.sf.jsqlparser.util.validation.Validation;
+import net.sf.jsqlparser.util.validation.feature.FeaturesAllowed;
+import net.sf.jsqlparser.util.validation.feature.SqlServerVersion;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class InsertBulkTest {
+    private static InsertBulk parse(String sql) throws Exception {
+        return (InsertBulk) CCJSqlParserUtil.parse(sql, p -> p.withDialect(Dialect.SQLSERVER));
+    }
+
+    @Test
+    void parsesIssue2033() throws Exception {
+        String sql = "INSERT BULK tpch.dbo.order_line ([ol_o_id] int, [ol_d_id] tinyint, "
+                + "[ol_w_id] int, [ol_number] tinyint, [ol_i_id] int, [ol_delivery_d] datetime, "
+                + "[ol_amount] smallmoney, [ol_supply_w_id] int, [ol_quantity] smallint, "
+                + "[ol_dist_info] char(24) COLLATE Chinese_PRC_CI_AS) WITH (ROWS_PER_BATCH = 500000)";
+        InsertBulk statement = (InsertBulk) assertSqlCanBeParsedAndDeparsed(sql, true,
+                p -> p.withDialect(Dialect.SQLSERVER));
+        assertEquals(10, statement.getColumns().size());
+        assertEquals("char(24)",
+                statement.getColumns().get(9).getColDataType().toString().replace(" ", ""));
+        assertEquals(List.of("COLLATE", "Chinese_PRC_CI_AS"),
+                statement.getColumns().get(9).getColumnSpecs());
+        assertEquals(InsertBulk.Option.Kind.ROWS_PER_BATCH,
+                statement.getOptions().get(0).getKind());
+        assertEquals("500000", statement.getOptions().get(0).getValue().toString());
+        assertEquals(Set.of("tpch.dbo.order_line"), new TablesNamesFinder().getTables(statement));
+        assertTrue(statement.getFeatures().modifiesData());
+        assertFalse(statement.getFeatures().modifiesSchema());
+        assertFalse(statement.getFeatures().returnsResultSet());
+        assertEquals(statement.toString(), parse(statement.toString()).toString());
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"", " WITH (TABLOCK)",
+            " WITH (CHECK_CONSTRAINTS, FIRE_TRIGGERS, KEEP_NULLS, ALLOW_ENCRYPTED_VALUE_MODIFICATIONS)",
+            " WITH (ORDER ([id] ASC, [name] DESC), ROWS_PER_BATCH = 100)"})
+    void supportsDriverOptionsAndColumnTypes(String options) throws Exception {
+        String sql = "INSERT BULK [db].[dbo].[items] ([id] decimal(18, 2), [name] nvarchar(max))"
+                + options;
+        InsertBulk statement = (InsertBulk) assertSqlCanBeParsedAndDeparsed(sql, true,
+                p -> p.withDialect(Dialect.SQLSERVER));
+        assertEquals(statement.toString(), parse(statement.toString()).toString());
+    }
+
+    @Test
+    void exposesMutableOptionsToVisitorsAndDeparsers() throws Exception {
+        InsertBulk statement =
+                parse("INSERT BULK t (id int) WITH (ROWS_PER_BATCH = 100, ORDER (id ASC))");
+        List<Long> values = new ArrayList<>();
+        ExpressionVisitorAdapter<Void> expressions = new ExpressionVisitorAdapter<Void>() {
+            @Override
+            public <S> Void visit(LongValue value, S context) {
+                assertEquals("context", context);
+                values.add(value.getValue());
+                return null;
+            }
+        };
+        statement.accept(new StatementVisitorAdapter<>(new SelectVisitorAdapter<>(expressions)),
+                "context");
+        assertEquals(List.of(100L), values);
+        statement.getColumns().get(0).getColDataType().setDataType("bigint");
+        statement.getOptions().get(0).setValue(new LongValue(200));
+        statement.getOptions().get(1).getOrderByElements().get(0).setAsc(false);
+        assertEquals("INSERT BULK t (id bigint) WITH (ROWS_PER_BATCH = 200, ORDER (id DESC))",
+                statement.toString());
+        StringBuilder output = new StringBuilder();
+        ExpressionDeParser deparser = new ExpressionDeParser() {
+            @Override
+            public <S> StringBuilder visit(LongValue value, S context) {
+                return getBuilder().append(value.getValue() + 1);
+            }
+        };
+        statement.accept(new StatementDeParser(deparser, new SelectDeParser(), output));
+        assertEquals(statement.toString().replace("200", "201"), output.toString());
+        assertEquals(output.toString(), parse(output.toString()).toString());
+    }
+
+    @Test
+    void gatesDialectAndRetainsOrdinaryInserts() throws Exception {
+        String sql = "INSERT BULK t (id int)";
+        assertThrows(JSQLParserException.class, () -> CCJSqlParserUtil.parse(sql));
+        for (Dialect dialect : Dialect.values()) {
+            if (dialect != Dialect.SQLSERVER) {
+                assertThrows(JSQLParserException.class,
+                        () -> CCJSqlParserUtil.parse(sql, p -> p.withDialect(dialect)));
+            }
+        }
+        assertSqlCanBeParsedAndDeparsed("INSERT INTO bulk (id) VALUES (1)", true,
+                p -> p.withDialect(Dialect.SQLSERVER));
+        assertSqlCanBeParsedAndDeparsed("INSERT INTO [bulk] (id) VALUES (1)", true,
+                p -> p.withDialect(Dialect.SQLSERVER));
+        assertEquals(2, CCJSqlParserUtil.parseStatements(sql + "; SELECT 1;",
+                p -> p.withDialect(Dialect.SQLSERVER)).size());
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"INSERT BULK t ()", "INSERT BULK t (id)",
+            "INSERT BULK t (id int) VALUES (1)",
+            "INSERT BULK t (id int) WITH ()", "INSERT BULK t (id int) WITH (UNKNOWN_OPTION)",
+            "INSERT BULK t (id int) WITH (TABLOCK = 1)",
+            "INSERT BULK t (id int) WITH (TABLOCK, TABLOCK)",
+            "INSERT BULK t (id int) WITH (ROWS_PER_BATCH)",
+            "INSERT BULK t (id int) WITH (ORDER ())"})
+    void rejectsMalformedBulkDeclarations(String sql) {
+        assertThrows(JSQLParserException.class, () -> parse(sql));
+    }
+
+    @Test
+    void validatesBulkCapability() {
+        FeatureConfiguration config = CCJSqlParserUtil.newParser("SELECT 1")
+                .withDialect(Dialect.SQLSERVER).getConfiguration();
+        assertTrue(new Validation(config, List.of(SqlServerVersion.values()[0]),
+                "INSERT BULK t (id int)")
+                .validate().isEmpty());
+        assertFalse(new Validation(config, List.of(new FeaturesAllowed(Feature.insert)),
+                "INSERT BULK t (id int)").validate().isEmpty());
+    }
+}