ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject [35/50] [abbrv] ignite git commit: IGNITE-6917: Implemented SQL COPY command
Date Fri, 09 Feb 2018 13:02:50 GMT
IGNITE-6917: Implemented SQL COPY command

This closes #3419


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/25d38cc9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/25d38cc9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/25d38cc9

Branch: refs/heads/ignite-7485-2
Commit: 25d38cc98c4ca098679d69ad90fc8bed66e6916d
Parents: faf50f1
Author: gg-shq <kshirokov@gridgain.com>
Authored: Wed Feb 7 14:28:04 2018 +0300
Committer: Igor Sapego <isapego@gridgain.com>
Committed: Wed Feb 7 14:30:39 2018 +0300

----------------------------------------------------------------------
 .../internal/jdbc2/JdbcBulkLoadSelfTest.java    | 185 ++++++
 .../ignite/jdbc/JdbcErrorsAbstractSelfTest.java |   2 +-
 .../jdbc/suite/IgniteJdbcDriverTestSuite.java   |  14 +
 .../thin/JdbcThinBulkLoadAbstractSelfTest.java  | 601 +++++++++++++++++++
 ...inBulkLoadAtomicPartitionedNearSelfTest.java |  39 ++
 ...bcThinBulkLoadAtomicPartitionedSelfTest.java |  39 ++
 ...dbcThinBulkLoadAtomicReplicatedSelfTest.java |  39 ++
 ...oadTransactionalPartitionedNearSelfTest.java |  39 ++
 ...ulkLoadTransactionalPartitionedSelfTest.java |  39 ++
 ...BulkLoadTransactionalReplicatedSelfTest.java |  39 ++
 .../JdbcThinDynamicIndexAbstractSelfTest.java   |   1 -
 .../clients/src/test/resources/bulkload0.csv    |   0
 .../clients/src/test/resources/bulkload1.csv    |   1 +
 .../clients/src/test/resources/bulkload2.csv    |   2 +
 .../src/test/resources/bulkload2_utf.csv        |   2 +
 .../cache/query/BulkLoadContextCursor.java      |  97 +++
 .../internal/jdbc/thin/JdbcThinStatement.java   |  68 ++-
 .../ignite/internal/jdbc2/JdbcQueryTask.java    |  12 +-
 .../bulkload/BulkLoadAckClientParameters.java   |  92 +++
 .../bulkload/BulkLoadCacheWriter.java           |  31 +
 .../processors/bulkload/BulkLoadCsvFormat.java  | 159 +++++
 .../processors/bulkload/BulkLoadCsvParser.java  |  65 ++
 .../processors/bulkload/BulkLoadFormat.java     |  33 +
 .../processors/bulkload/BulkLoadParser.java     |  61 ++
 .../processors/bulkload/BulkLoadProcessor.java  | 104 ++++
 .../bulkload/BulkLoadStreamerWriter.java        |  65 ++
 .../bulkload/pipeline/CharsetDecoderBlock.java  | 132 ++++
 .../pipeline/CsvLineProcessorBlock.java         |  70 +++
 .../bulkload/pipeline/LineSplitterBlock.java    |  72 +++
 .../bulkload/pipeline/PipelineBlock.java        |  66 ++
 .../bulkload/pipeline/StrListAppenderBlock.java |  52 ++
 .../odbc/jdbc/JdbcBulkLoadAckResult.java        | 111 ++++
 .../odbc/jdbc/JdbcBulkLoadBatchRequest.java     | 183 ++++++
 .../odbc/jdbc/JdbcBulkLoadProcessor.java        | 144 +++++
 .../processors/odbc/jdbc/JdbcRequest.java       |   7 +
 .../odbc/jdbc/JdbcRequestHandler.java           |  90 ++-
 .../processors/odbc/jdbc/JdbcResult.java        |   8 +
 .../apache/ignite/internal/sql/SqlKeyword.java  |  15 +
 .../apache/ignite/internal/sql/SqlParser.java   |  18 +-
 .../sql/command/SqlBulkLoadCommand.java         | 273 +++++++++
 .../internal/sql/SqlParserBulkLoadSelfTest.java |  70 +++
 .../query/h2/DmlStatementsProcessor.java        |  99 +++
 .../processors/query/h2/IgniteH2Indexing.java   |  35 +-
 .../query/h2/ddl/DdlStatementsProcessor.java    |   2 +
 .../processors/query/h2/dml/UpdateMode.java     |  11 +-
 .../processors/query/h2/dml/UpdatePlan.java     |  20 +-
 .../query/h2/dml/UpdatePlanBuilder.java         |  86 +++
 .../IgniteCacheQuerySelfTestSuite.java          |   2 +
 parent/pom.xml                                  |   3 +-
 49 files changed, 3361 insertions(+), 37 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/internal/jdbc2/JdbcBulkLoadSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/internal/jdbc2/JdbcBulkLoadSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/internal/jdbc2/JdbcBulkLoadSelfTest.java
new file mode 100644
index 0000000..d9506cf
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/internal/jdbc2/JdbcBulkLoadSelfTest.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.jdbc2;
+
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import java.io.Serializable;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+
+import static org.apache.ignite.IgniteJdbcDriver.CFG_URL_PREFIX;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/** COPY command test for the regular JDBC driver. */
+public class JdbcBulkLoadSelfTest extends GridCommonAbstractTest {
+    /** JDBC URL. */
+    private static final String BASE_URL = CFG_URL_PREFIX +
+        "cache=default@modules/clients/src/test/config/jdbc-config.xml";
+
+    /** Connection. */
+    protected Connection conn;
+
+    /** The logger. */
+    protected transient IgniteLogger log;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        return getConfiguration0(gridName);
+    }
+
+    /**
+     * @param gridName Grid name.
+     * @return Grid configuration used for starting the grid.
+     * @throws Exception If failed.
+     */
+    private IgniteConfiguration getConfiguration0(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        CacheConfiguration<?,?> cache = defaultCacheConfiguration();
+
+        cache.setCacheMode(PARTITIONED);
+        cache.setBackups(1);
+        cache.setWriteSynchronizationMode(FULL_SYNC);
+        cache.setIndexedTypes(
+            Integer.class, Person.class
+        );
+
+        cfg.setCacheConfiguration(cache);
+        cfg.setLocalHost("127.0.0.1");
+
+        TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+        TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+        ipFinder.setAddresses(Collections.singleton("127.0.0.1:47500..47501"));
+
+        disco.setIpFinder(ipFinder);
+
+        cfg.setDiscoverySpi(disco);
+
+        cfg.setConnectorConfiguration(new ConnectorConfiguration());
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        startGrids(2);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        stopAllGrids();
+    }
+
+    /**
+     * Establishes the JDBC connection.
+     *
+     * @return Connection to use for the test.
+     * @throws Exception if failed.
+     */
+    private Connection createConnection() throws Exception {
+        Properties props = new Properties();
+
+        return DriverManager.getConnection(BASE_URL, props);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        U.closeQuiet(conn);
+
+        ignite(0).cache(DEFAULT_CACHE_NAME).clear();
+
+        super.afterTest();
+    }
+
+    /**
+     * This is more a placeholder for implementation of IGNITE-7553.
+     *
+     * @throws Exception if failed.
+     */
+    public void testBulkLoadThrows() throws Exception {
+        GridTestUtils.assertThrows(null, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                conn = createConnection();
+
+                try (Statement stmt = conn.createStatement()) {
+                    stmt.executeUpdate("copy from \"dummy.csv\" into Person" +
+                        " (_key, id, firstName, lastName) format csv");
+
+                    return null;
+                }
+            }
+        }, SQLException.class, "COPY command is currently supported only in thin JDBC driver.");
+    }
+
+    /**
+     * A test class for creating a query entity.
+     */
+    @SuppressWarnings("UnusedDeclaration")
+    private static class Person implements Serializable {
+        /** ID. */
+        @QuerySqlField
+        private final int id;
+
+        /** First name. */
+        @QuerySqlField(index = false)
+        private final String firstName;
+
+        /** Last name. */
+        @QuerySqlField(index = false)
+        private final String lastName;
+
+        /** Age. */
+        @QuerySqlField
+        private final int age;
+
+        /**
+         * @param id ID.
+         * @param firstName First name
+         * @param lastName Last name
+         * @param age Age.
+         */
+        private Person(int id, String firstName, String lastName, int age) {
+            assert !F.isEmpty(firstName);
+            assert !F.isEmpty(lastName);
+            assert age > 0;
+
+            this.id = id;
+            this.firstName = firstName;
+            this.lastName = lastName;
+            this.age = age;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcErrorsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcErrorsAbstractSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcErrorsAbstractSelfTest.java
index 49746b6..2059408 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcErrorsAbstractSelfTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcErrorsAbstractSelfTest.java
@@ -107,7 +107,7 @@ public abstract class JdbcErrorsAbstractSelfTest extends GridCommonAbstractTest
      */
     public void testDmlErrors() throws SQLException {
         checkErrorState("INSERT INTO \"test\".INTEGER(_key, _val) values(1, null)", "22004",
-            "Value for INSERT, MERGE, or UPDATE must not be null");
+            "Value for INSERT, COPY, MERGE, or UPDATE must not be null");
 
         checkErrorState("INSERT INTO \"test\".INTEGER(_key, _val) values(1, 'zzz')", "0700B",
             "Value conversion failed [from=java.lang.String, to=java.lang.Integer]");

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
index ff4d69f..656e218 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
@@ -35,6 +35,12 @@ import org.apache.ignite.jdbc.JdbcResultSetSelfTest;
 import org.apache.ignite.jdbc.JdbcStatementSelfTest;
 import org.apache.ignite.jdbc.thin.JdbcThinAutoCloseServerCursorTest;
 import org.apache.ignite.jdbc.thin.JdbcThinBatchSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadAtomicPartitionedNearSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadAtomicPartitionedSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadAtomicReplicatedSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadTransactionalPartitionedNearSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadTransactionalPartitionedSelfTest;
+import org.apache.ignite.jdbc.thin.JdbcThinBulkLoadTransactionalReplicatedSelfTest;
 import org.apache.ignite.jdbc.thin.JdbcThinComplexDmlDdlSelfTest;
 import org.apache.ignite.jdbc.thin.JdbcThinComplexQuerySelfTest;
 import org.apache.ignite.jdbc.thin.JdbcThinConnectionSelfTest;
@@ -154,6 +160,14 @@ public class IgniteJdbcDriverTestSuite extends TestSuite {
         suite.addTest(new TestSuite(JdbcThinDynamicIndexTransactionalPartitionedSelfTest.class));
         suite.addTest(new TestSuite(JdbcThinDynamicIndexTransactionalReplicatedSelfTest.class));
 
+        // New thin JDBC driver, DML tests
+        suite.addTest(new TestSuite(JdbcThinBulkLoadAtomicPartitionedNearSelfTest.class));
+        suite.addTest(new TestSuite(JdbcThinBulkLoadAtomicPartitionedSelfTest.class));
+        suite.addTest(new TestSuite(JdbcThinBulkLoadAtomicReplicatedSelfTest.class));
+        suite.addTest(new TestSuite(JdbcThinBulkLoadTransactionalPartitionedNearSelfTest.class));
+        suite.addTest(new TestSuite(JdbcThinBulkLoadTransactionalPartitionedSelfTest.class));
+        suite.addTest(new TestSuite(JdbcThinBulkLoadTransactionalReplicatedSelfTest.class));
+
         // New thin JDBC driver, full SQL tests
         suite.addTest(new TestSuite(JdbcThinComplexDmlDdlSelfTest.class));
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAbstractSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAbstractSelfTest.java
new file mode 100644
index 0000000..761f700
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAbstractSelfTest.java
@@ -0,0 +1,601 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.testframework.GridTestUtils;
+
+import java.sql.BatchUpdateException;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.concurrent.Callable;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_PARSER_DISABLE_H2_FALLBACK;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath;
+
+/**
+ * COPY statement tests.
+ */
+public abstract class JdbcThinBulkLoadAbstractSelfTest extends JdbcThinAbstractDmlStatementSelfTest {
+    /** Default table name. */
+    private static final String TBL_NAME = "Person";
+
+    /** JDBC statement. */
+    private Statement stmt;
+
+    /** A CSV file with zero records */
+    private static final String BULKLOAD_EMPTY_CSV_FILE =
+        Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload0.csv"))
+            .getAbsolutePath();
+
+    /** A CSV file with one record. */
+    private static final String BULKLOAD_ONE_LINE_CSV_FILE =
+        Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload1.csv"))
+            .getAbsolutePath();
+
+    /** A CSV file with two records. */
+    private static final String BULKLOAD_TWO_LINES_CSV_FILE =
+        Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload2.csv"))
+            .getAbsolutePath();
+
+    /** A file with UTF records. */
+    private static final String BULKLOAD_UTF_CSV_FILE =
+        Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload2_utf.csv"))
+            .getAbsolutePath();
+
+    /** Basic COPY statement used in majority of the tests. */
+    public static final String BASIC_SQL_COPY_STMT =
+        "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\"" +
+            " into " + TBL_NAME +
+            " (_key, age, firstName, lastName)" +
+            " format csv";
+
+    /** {@inheritDoc} */
+    @Override protected CacheConfiguration cacheConfig() {
+        return cacheConfigWithIndexedTypes();
+    }
+
+    /**
+     * Creates cache configuration with {@link QueryEntity} created
+     * using {@link CacheConfiguration#setIndexedTypes(Class[])} call.
+     *
+     * @return The cache configuration.
+     */
+    @SuppressWarnings("unchecked")
+    private CacheConfiguration cacheConfigWithIndexedTypes() {
+        CacheConfiguration<?,?> cache = defaultCacheConfiguration();
+
+        cache.setCacheMode(cacheMode());
+        cache.setAtomicityMode(atomicityMode());
+        cache.setWriteSynchronizationMode(FULL_SYNC);
+
+        if (cacheMode() == PARTITIONED)
+            cache.setBackups(1);
+
+        if (nearCache())
+            cache.setNearConfiguration(new NearCacheConfiguration());
+
+        cache.setIndexedTypes(
+            String.class, Person.class
+        );
+
+        return cache;
+    }
+
+    /**
+     * Returns true if we are testing near cache.
+     *
+     * @return true if we are testing near cache.
+     */
+    protected abstract boolean nearCache();
+
+    /**
+     * Returns cache atomicity mode we are testing.
+     *
+     * @return The cache atomicity mode we are testing.
+     */
+    protected abstract CacheAtomicityMode atomicityMode();
+
+    /**
+     * Returns cache mode we are testing.
+     *
+     * @return The cache mode we are testing.
+     */
+    protected abstract CacheMode cacheMode();
+
+    /**
+     * Creates cache configuration with {@link QueryEntity} created
+     * using {@link CacheConfiguration#setQueryEntities(Collection)} call.
+     *
+     * @return The cache configuration.
+     */
+    private CacheConfiguration cacheConfigWithQueryEntity() {
+        CacheConfiguration<?,?> cache = defaultCacheConfiguration();
+
+        cache.setCacheMode(PARTITIONED);
+        cache.setBackups(1);
+        cache.setWriteSynchronizationMode(FULL_SYNC);
+
+        QueryEntity e = new QueryEntity();
+
+        e.setKeyType(String.class.getName());
+        e.setValueType("Person");
+
+        e.addQueryField("id", Integer.class.getName(), null);
+        e.addQueryField("age", Integer.class.getName(), null);
+        e.addQueryField("firstName", String.class.getName(), null);
+        e.addQueryField("lastName", String.class.getName(), null);
+
+        cache.setQueryEntities(Collections.singletonList(e));
+
+        return cache;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+
+        System.setProperty(IGNITE_SQL_PARSER_DISABLE_H2_FALLBACK, "TRUE");
+
+        stmt = conn.createStatement();
+
+        assertNotNull(stmt);
+        assertFalse(stmt.isClosed());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        if (stmt != null && !stmt.isClosed())
+            stmt.close();
+
+        assertTrue(stmt.isClosed());
+
+        System.clearProperty(IGNITE_SQL_PARSER_DISABLE_H2_FALLBACK);
+
+        super.afterTest();
+    }
+
+    /**
+     * Dead-on-arrival test. Imports two-entry CSV file into a table and checks
+     * the created entries using SELECT statement.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testBasicStatement() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(BASIC_SQL_COPY_STMT);
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Imports two-entry CSV file with UTF-8 characters into a table and checks
+     * the created entries using SELECT statement.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testUtf() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_UTF_CSV_FILE + "\" into " + TBL_NAME +
+                " (_key, age, firstName, lastName)" +
+                " format csv");
+
+        assertEquals(2, updatesCnt);
+
+        checkUtfCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Imports two-entry CSV file with UTF-8 characters into a table using batch size of one byte
+     * (thus splitting each two-byte UTF-8 character into two batches)
+     * and checks the created entries using SELECT statement.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testUtfBatchSize_1() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_UTF_CSV_FILE + "\" into " + TBL_NAME +
+                " (_key, age, firstName, lastName)" +
+                " format csv batch_size 1");
+
+        assertEquals(2, updatesCnt);
+
+        checkUtfCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Imports one-entry CSV file into a table and checks the entry created using SELECT statement.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testOneLineFile() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_ONE_LINE_CSV_FILE + "\" into " + TBL_NAME +
+                " (_key, age, firstName, lastName)" +
+                " format csv");
+
+        assertEquals(1, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 1);
+    }
+
+    /**
+     * Imports zero-entry CSV file into a table and checks that no entries are created
+     * using SELECT statement.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testEmptyFile() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_EMPTY_CSV_FILE + "\" into " + TBL_NAME +
+                " (_key, age, firstName, lastName)" +
+                " format csv");
+
+        assertEquals(0, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 0);
+    }
+
+    /**
+     * Checks that error is reported for a non-existent file.
+     */
+    public void testWrongFileName() {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.executeUpdate(
+                    "copy from \"nonexistent\" into Person" +
+                        " (_key, age, firstName, lastName)" +
+                        " format csv");
+
+                return null;
+            }
+        }, SQLException.class, "Failed to read file: 'nonexistent'");
+    }
+
+    /**
+     * Checks that error is reported if the destination table is missing.
+     */
+    public void testMissingTable() {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.executeUpdate(
+                    "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into Peterson" +
+                        " (_key, age, firstName, lastName)" +
+                        " format csv");
+
+                return null;
+            }
+        }, SQLException.class, "Table does not exist: PETERSON");
+    }
+
+    /**
+     * Checks that error is reported when a non-existing column is specified in the SQL command.
+     */
+    public void testWrongColumnName() {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.executeUpdate(
+                    "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into Person" +
+                        " (_key, age, firstName, lostName)" +
+                        " format csv");
+
+                return null;
+            }
+        }, SQLException.class, "Column \"LOSTNAME\" not found");
+    }
+
+    /**
+     * Checks that error is reported if field read from CSV file cannot be converted to the type of the column.
+     */
+    public void testWrongColumnType() {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.executeUpdate(
+                    "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into Person" +
+                        " (_key, firstName, age, lastName)" +
+                        " format csv");
+
+                return null;
+            }
+        }, SQLException.class, "Value conversion failed [from=java.lang.String, to=java.lang.Integer]");
+    }
+
+    /**
+     * Checks that if even a subset of fields is imported, the imported fields are set correctly.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testFieldsSubset() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into " + TBL_NAME +
+                " (_key, age, firstName)" +
+                " format csv");
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(TBL_NAME, false, 2);
+    }
+
+    /**
+     * Checks that bulk load works when we create table using 'CREATE TABLE' command.
+     *
+     * The majority of the tests in this class use {@link CacheConfiguration#setIndexedTypes(Class[])}
+     * to create the table.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testCreateAndBulkLoadTable() throws SQLException {
+        String tblName = QueryUtils.DFLT_SCHEMA + ".\"PersonTbl\"";
+
+        execute(conn, "create table " + tblName +
+            " (id int primary key, age int, firstName varchar(30), lastName varchar(30))");
+
+        int updatesCnt = stmt.executeUpdate(
+            "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into " + tblName +
+                "(_key, age, firstName, lastName)" +
+                " format csv");
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(tblName, true, 2);
+    }
+
+    /**
+     * Checks that bulk load works when we create table with {@link CacheConfiguration#setQueryEntities(Collection)}.
+     *
+     * The majority of the tests in this class use {@link CacheConfiguration#setIndexedTypes(Class[])}
+     * to create a table.
+     *
+     * @throws SQLException If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void testConfigureQueryEntityAndBulkLoad() throws SQLException {
+        ignite(0).getOrCreateCache(cacheConfigWithQueryEntity());
+
+        int updatesCnt = stmt.executeUpdate(BASIC_SQL_COPY_STMT);
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Checks that bulk load works when we use batch size of 1 byte and thus
+     * create multiple batches per COPY.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testBatchSize_1() throws SQLException {
+        int updatesCnt = stmt.executeUpdate(BASIC_SQL_COPY_STMT + " batch_size 1");
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Verifies exception thrown if COPY is added into a batch.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testMultipleStatement() throws SQLException {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.addBatch(BASIC_SQL_COPY_STMT);
+
+                stmt.addBatch("copy from \"" + BULKLOAD_ONE_LINE_CSV_FILE + "\" into " + TBL_NAME +
+                    " (_key, age, firstName, lastName)" +
+                    " format csv");
+
+                stmt.addBatch("copy from \"" + BULKLOAD_UTF_CSV_FILE + "\" into " + TBL_NAME +
+                    " (_key, age, firstName, lastName)" +
+                    " format csv");
+
+                stmt.executeBatch();
+
+                return null;
+            }
+        }, BatchUpdateException.class, "COPY command cannot be executed in batch mode.");
+    }
+
+    /**
+     * Verifies that COPY command is rejected by Statement.executeQuery().
+     *
+     * @throws SQLException If failed.
+     */
+    public void testExecuteQuery() throws SQLException {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                stmt.executeQuery(BASIC_SQL_COPY_STMT);
+
+                return null;
+            }
+        }, SQLException.class, "The query isn't SELECT query");
+    }
+
+    /**
+     * Verifies that COPY command works in Statement.execute().
+     *
+     * @throws SQLException If failed.
+     */
+    public void testExecute() throws SQLException {
+        boolean isRowSet = stmt.execute(BASIC_SQL_COPY_STMT);
+
+        assertFalse(isRowSet);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Verifies that COPY command can be called with PreparedStatement.executeUpdate().
+     *
+     * @throws SQLException If failed.
+     */
+    public void testPreparedStatementWithExecuteUpdate() throws SQLException {
+        PreparedStatement pstmt = conn.prepareStatement(BASIC_SQL_COPY_STMT);
+
+        int updatesCnt = pstmt.executeUpdate();
+
+        assertEquals(2, updatesCnt);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Verifies that COPY command reports an error when used with PreparedStatement parameter.
+     *
+     * @throws SQLException If failed.
+     */
+    public void testPreparedStatementWithParameter() throws SQLException {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+                @Override public Object call() throws Exception {
+                    PreparedStatement pstmt = conn.prepareStatement(
+                        "copy from \"" + BULKLOAD_TWO_LINES_CSV_FILE + "\" into " + TBL_NAME +
+                            " (_key, age, firstName, lastName)" +
+                            " format ?");
+
+                    pstmt.setString(1, "csv");
+
+                    pstmt.executeUpdate();
+
+                    return null;
+                }
+            }, SQLException.class, "Unexpected token: \"?\" (expected: \"[identifier]\"");
+    }
+
+    /**
+     * Verifies that COPY command can be called with PreparedStatement.execute().
+     *
+     * @throws SQLException If failed.
+     */
+    public void testPreparedStatementWithExecute() throws SQLException {
+        PreparedStatement pstmt = conn.prepareStatement(BASIC_SQL_COPY_STMT);
+
+        boolean isRowSet = pstmt.execute();
+
+        assertFalse(isRowSet);
+
+        checkCacheContents(TBL_NAME, true, 2);
+    }
+
+    /**
+     * Verifies that COPY command is rejected by PreparedStatement.executeQuery().
+     */
+    public void testPreparedStatementWithExecuteQuery() {
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                PreparedStatement pstmt = conn.prepareStatement(BASIC_SQL_COPY_STMT);
+
+                pstmt.executeQuery();
+
+                return null;
+            }
+        }, SQLException.class, "The query isn't SELECT query");
+    }
+
+    /**
+     * Checks cache contents for a typical test using SQL SELECT command.
+     *
+     * @param tblName Table name to query.
+     * @param checkLastName Check 'lastName' column (not imported in some tests).
+     * @param recCnt Number of records to expect.
+     * @throws SQLException When one of checks has failed.
+     */
+    private void checkCacheContents(String tblName, boolean checkLastName, int recCnt) throws SQLException {
+        ResultSet rs = stmt.executeQuery("select _key, age, firstName, lastName from " + tblName);
+
+        assert rs != null;
+
+        int cnt = 0;
+
+        while (rs.next()) {
+            int id = rs.getInt("_key");
+
+            if (id == 123) {
+                assertEquals(12, rs.getInt("age"));
+                assertEquals("FirstName123 MiddleName123", rs.getString("firstName"));
+                if (checkLastName)
+                    assertEquals("LastName123", rs.getString("lastName"));
+            }
+            else if (id == 456) {
+                assertEquals(45, rs.getInt("age"));
+                assertEquals("FirstName456", rs.getString("firstName"));
+                if (checkLastName)
+                    assertEquals("LastName456", rs.getString("lastName"));
+            }
+            else
+                fail("Wrong ID: " + id);
+
+            cnt++;
+        }
+
+        assertEquals(recCnt, cnt);
+    }
+
+    /**
+     * Checks cache contents for a UTF-8 bulk load tests using SQL SELECT command.
+     *
+     * @param tblName Table name to query.
+     * @param checkLastName Check 'lastName' column (not imported in some tests).
+     * @param recCnt Number of records to expect.
+     * @throws SQLException When one of checks has failed.
+     */
+    private void checkUtfCacheContents(String tblName, boolean checkLastName, int recCnt) throws SQLException {
+        ResultSet rs = stmt.executeQuery("select _key, age, firstName, lastName from " + tblName);
+
+        assert rs != null;
+
+        int cnt = 0;
+
+        while (rs.next()) {
+            int id = rs.getInt("_key");
+
+            if (id == 123) {
+                assertEquals(12, rs.getInt("age"));
+                assertEquals("Имя123 Отчество123", rs.getString("firstName"));
+                if (checkLastName)
+                    assertEquals("Фамилия123", rs.getString("lastName"));
+            }
+            else if (id == 456) {
+                assertEquals(45, rs.getInt("age"));
+                assertEquals("Имя456", rs.getString("firstName"));
+                if (checkLastName)
+                    assertEquals("Фамилия456", rs.getString("lastName"));
+            }
+            else
+                fail("Wrong ID: " + id);
+
+            cnt++;
+        }
+
+        assertEquals(recCnt, cnt);
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedNearSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedNearSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedNearSelfTest.java
new file mode 100644
index 0000000..887b1d9
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedNearSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for partitioned atomic near-cache mode. */
+public class JdbcThinBulkLoadAtomicPartitionedNearSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.ATOMIC;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return true;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedSelfTest.java
new file mode 100644
index 0000000..5581333
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicPartitionedSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for partitioned atomic mode. */
+public class JdbcThinBulkLoadAtomicPartitionedSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.ATOMIC;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicReplicatedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicReplicatedSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicReplicatedSelfTest.java
new file mode 100644
index 0000000..c3d69af
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadAtomicReplicatedSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for replicated atomic near-cache mode. */
+public class JdbcThinBulkLoadAtomicReplicatedSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.REPLICATED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.ATOMIC;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedNearSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedNearSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedNearSelfTest.java
new file mode 100644
index 0000000..9336dd1
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedNearSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for partitioned transactional near-cache mode. */
+public class JdbcThinBulkLoadTransactionalPartitionedNearSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.TRANSACTIONAL;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return true;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedSelfTest.java
new file mode 100644
index 0000000..d1dea2a
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalPartitionedSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for partitioned transactional mode. */
+public class JdbcThinBulkLoadTransactionalPartitionedSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.TRANSACTIONAL;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalReplicatedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalReplicatedSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalReplicatedSelfTest.java
new file mode 100644
index 0000000..1c377fa
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinBulkLoadTransactionalReplicatedSelfTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.jdbc.thin;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+/** A {@link JdbcThinBulkLoadAbstractSelfTest} for replicated transactional mode. */
+public class JdbcThinBulkLoadTransactionalReplicatedSelfTest extends JdbcThinBulkLoadAbstractSelfTest {
+    /** {@inheritDoc} */
+    @Override protected CacheMode cacheMode() {
+        return CacheMode.REPLICATED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheAtomicityMode atomicityMode() {
+        return CacheAtomicityMode.TRANSACTIONAL;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean nearCache() {
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinDynamicIndexAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinDynamicIndexAbstractSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinDynamicIndexAbstractSelfTest.java
index dbe93a4..539713a 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinDynamicIndexAbstractSelfTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinDynamicIndexAbstractSelfTest.java
@@ -25,7 +25,6 @@ import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Callable;
 import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cache.CacheAtomicityMode;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.CacheWriteSynchronizationMode;

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/resources/bulkload0.csv
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/resources/bulkload0.csv b/modules/clients/src/test/resources/bulkload0.csv
new file mode 100644
index 0000000..e69de29

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/resources/bulkload1.csv
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/resources/bulkload1.csv b/modules/clients/src/test/resources/bulkload1.csv
new file mode 100644
index 0000000..596ac32
--- /dev/null
+++ b/modules/clients/src/test/resources/bulkload1.csv
@@ -0,0 +1 @@
+123,12,"FirstName123 MiddleName123",LastName123
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/resources/bulkload2.csv
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/resources/bulkload2.csv b/modules/clients/src/test/resources/bulkload2.csv
new file mode 100644
index 0000000..d398c19
--- /dev/null
+++ b/modules/clients/src/test/resources/bulkload2.csv
@@ -0,0 +1,2 @@
+123,12,"FirstName123 MiddleName123",LastName123
+456,45,"FirstName456","LastName456"
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/clients/src/test/resources/bulkload2_utf.csv
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/resources/bulkload2_utf.csv b/modules/clients/src/test/resources/bulkload2_utf.csv
new file mode 100644
index 0000000..bdb6489
--- /dev/null
+++ b/modules/clients/src/test/resources/bulkload2_utf.csv
@@ -0,0 +1,2 @@
+123,12,"Имя123 Отчество123",Фамилия123
+456,45,"Имя456","Фамилия456"
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/cache/query/BulkLoadContextCursor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/BulkLoadContextCursor.java b/modules/core/src/main/java/org/apache/ignite/cache/query/BulkLoadContextCursor.java
new file mode 100644
index 0000000..b7fdec3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/BulkLoadContextCursor.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.query;
+
+import org.apache.ignite.internal.processors.bulkload.BulkLoadAckClientParameters;
+import org.apache.ignite.internal.processors.bulkload.BulkLoadProcessor;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A special FieldsQueryCursor subclass that is used as a sentinel to transfer data from bulk load
+ * (COPY) command to the JDBC or other client-facing driver: the bulk load batch processor
+ * and parameters to send to the client.
+ * */
+public class BulkLoadContextCursor implements FieldsQueryCursor<List<?>> {
+    /** Bulk load context from SQL command. */
+    private final BulkLoadProcessor processor;
+
+    /** Bulk load parameters to send to the client. */
+    private final BulkLoadAckClientParameters clientParams;
+
+    /**
+     * Creates a cursor.
+     *
+     * @param processor Bulk load context object to store.
+     * @param clientParams Parameters to send to client.
+     */
+    public BulkLoadContextCursor(BulkLoadProcessor processor, BulkLoadAckClientParameters clientParams) {
+        this.processor = processor;
+        this.clientParams = clientParams;
+    }
+
+    /**
+     * Returns a bulk load context.
+     *
+     * @return a bulk load context.
+     */
+    public BulkLoadProcessor bulkLoadProcessor() {
+        return processor;
+    }
+
+    /**
+     * Returns the bulk load parameters to send to the client.
+     *
+     * @return The bulk load parameters to send to the client.
+     */
+    public BulkLoadAckClientParameters clientParams() {
+        return clientParams;
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<List<?>> getAll() {
+        return Collections.singletonList(Arrays.asList(processor, clientParams));
+    }
+
+    /** {@inheritDoc} */
+    @NotNull @Override public Iterator<List<?>> iterator() {
+        return getAll().iterator();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void close() {
+        // no-op
+    }
+
+    /** {@inheritDoc} */
+    @Override public String getFieldName(int idx) {
+        if (idx < 0 || idx > 1)
+            throw new IndexOutOfBoundsException();
+
+        return idx == 0 ? "processor" : "clientParams";
+    }
+
+    /** {@inheritDoc} */
+    @Override public int getColumnsCount() {
+        return 2;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinStatement.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinStatement.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinStatement.java
index d29df93..2020011 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinStatement.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinStatement.java
@@ -17,6 +17,9 @@
 
 package org.apache.ignite.internal.jdbc.thin;
 
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.InputStream;
 import java.sql.BatchUpdateException;
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -25,21 +28,24 @@ import java.sql.SQLFeatureNotSupportedException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import org.apache.ignite.cache.query.SqlQuery;
 import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
-import org.apache.ignite.internal.processors.odbc.SqlStateCode;
 import org.apache.ignite.internal.processors.odbc.ClientListenerResponse;
+import org.apache.ignite.internal.processors.odbc.SqlStateCode;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcBatchExecuteRequest;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcBatchExecuteResult;
+import org.apache.ignite.internal.processors.odbc.jdbc.JdbcBulkLoadAckResult;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcQuery;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcQueryExecuteMultipleStatementsResult;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcQueryExecuteRequest;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcQueryExecuteResult;
-import org.apache.ignite.internal.processors.odbc.jdbc.JdbcStatementType;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcResult;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcResultInfo;
+import org.apache.ignite.internal.processors.odbc.jdbc.JdbcBulkLoadBatchRequest;
+import org.apache.ignite.internal.processors.odbc.jdbc.JdbcStatementType;
 
 import static java.sql.ResultSet.CONCUR_READ_ONLY;
 import static java.sql.ResultSet.FETCH_FORWARD;
@@ -132,6 +138,9 @@ public class JdbcThinStatement implements Statement {
 
         assert res0 != null;
 
+        if (res0 instanceof JdbcBulkLoadAckResult)
+            res0 = sendFile((JdbcBulkLoadAckResult)res0);
+
         if (res0 instanceof JdbcQueryExecuteResult) {
             JdbcQueryExecuteResult res = (JdbcQueryExecuteResult)res0;
 
@@ -176,6 +185,61 @@ public class JdbcThinStatement implements Statement {
         assert resultSets.size() > 0 : "At least one results set is expected";
     }
 
+    /**
+     * Sends a file to server in batches via multiple {@link JdbcBulkLoadBatchRequest}s.
+     *
+     * @param cmdRes Result of invoking COPY command: contains server-parsed
+     *    bulk load parameters, such as file name and batch size.
+     */
+    private JdbcResult sendFile(JdbcBulkLoadAckResult cmdRes) throws SQLException {
+        String fileName = cmdRes.params().localFileName();
+        int batchSize = cmdRes.params().batchSize();
+
+        int batchNum = 0;
+
+        try {
+            try (InputStream input = new BufferedInputStream(new FileInputStream(fileName))) {
+                byte[] buf = new byte[batchSize];
+
+                int readBytes;
+                while ((readBytes = input.read(buf)) != -1) {
+                    if (readBytes == 0)
+                        continue;
+
+                    JdbcResult res = conn.sendRequest(new JdbcBulkLoadBatchRequest(
+                        cmdRes.queryId(),
+                        batchNum++,
+                        JdbcBulkLoadBatchRequest.CMD_CONTINUE,
+                        readBytes == buf.length ? buf : Arrays.copyOf(buf, readBytes)));
+
+                    if (!(res instanceof JdbcQueryExecuteResult))
+                        throw new SQLException("Unknown response sent by the server: " + res);
+                }
+
+                return conn.sendRequest(new JdbcBulkLoadBatchRequest(
+                    cmdRes.queryId(),
+                    batchNum++,
+                    JdbcBulkLoadBatchRequest.CMD_FINISHED_EOF));
+            }
+        }
+        catch (Exception e) {
+            try {
+                conn.sendRequest(new JdbcBulkLoadBatchRequest(
+                    cmdRes.queryId(),
+                    batchNum,
+                    JdbcBulkLoadBatchRequest.CMD_FINISHED_ERROR));
+            }
+            catch (SQLException e1) {
+                throw new SQLException("Cannot send finalization request: " + e1.getMessage(), e);
+            }
+
+            if (e instanceof SQLException)
+                throw (SQLException) e;
+            else
+                throw new SQLException("Failed to read file: '" + fileName + "'", SqlStateCode.INTERNAL_ERROR, e);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public int executeUpdate(String sql) throws SQLException {
         execute0(JdbcStatementType.UPDATE_STMT_TYPE, sql, null);

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
index aa9f009..07034f4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
@@ -32,6 +32,8 @@ import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteJdbcDriver;
 import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.query.BulkLoadContextCursor;
+import org.apache.ignite.cache.query.FieldsQueryCursor;
 import org.apache.ignite.cache.query.QueryCursor;
 import org.apache.ignite.cache.query.SqlFieldsQuery;
 import org.apache.ignite.internal.IgniteKernal;
@@ -168,7 +170,15 @@ class JdbcQueryTask implements IgniteCallable<JdbcQueryTaskResult> {
             qry.setLazy(lazy());
             qry.setSchema(schemaName);
 
-            QueryCursorImpl<List<?>> qryCursor = (QueryCursorImpl<List<?>>)cache.withKeepBinary().query(qry);
+            FieldsQueryCursor<List<?>> fldQryCursor = cache.withKeepBinary().query(qry);
+
+            if (fldQryCursor instanceof BulkLoadContextCursor) {
+                fldQryCursor.close();
+                
+                throw new SQLException("COPY command is currently supported only in thin JDBC driver.");
+            }
+
+            QueryCursorImpl<List<?>> qryCursor = (QueryCursorImpl<List<?>>)fldQryCursor;
 
             if (isQry == null)
                 isQry = qryCursor.isQuery();

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadAckClientParameters.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadAckClientParameters.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadAckClientParameters.java
new file mode 100644
index 0000000..119d9f9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadAckClientParameters.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Bulk load parameters, which are parsed from SQL command and sent from server to client.
+ */
+public class BulkLoadAckClientParameters {
+    /** Minimum batch size. */
+    public static final int MIN_BATCH_SIZE = 1;
+
+    /**
+     * Maximum batch size. Note that the batch is wrapped to transport objects and the overall packet should fit
+     * into a Java array. 512 has been chosen arbitrarily.
+     */
+    public static final int MAX_BATCH_SIZE = Integer.MAX_VALUE - 512;
+
+    /** Size of a file batch for COPY command. */
+    public static final int DEFAULT_BATCH_SIZE = 4 * 1024 * 1024;
+
+    /** Local name of the file to send to server */
+    @NotNull private final String locFileName;
+
+    /** File batch size in bytes. */
+    private final int batchSize;
+
+    /**
+     * Creates a bulk load parameters.
+     *
+     * @param locFileName File name to send from client to server.
+     * @param batchSize Batch size (Number of bytes in a portion of a file to send in one JDBC request/response).
+     */
+    public BulkLoadAckClientParameters(@NotNull String locFileName, int batchSize) {
+        this.locFileName = locFileName;
+        this.batchSize = batchSize;
+    }
+
+    /**
+     * Returns the local name of file to send.
+     *
+     * @return The local name of file to send.
+     */
+    @NotNull public String localFileName() {
+        return locFileName;
+    }
+
+    /**
+     * Returns the batch size.
+     *
+     * @return The batch size.
+     */
+    public int batchSize() {
+        return batchSize;
+    }
+
+    /**
+     * Checks if batch size value is valid.
+     *
+     * @param sz The batch size to check.
+     * @throws IllegalArgumentException if batch size is invalid.
+     */
+    public static boolean isValidBatchSize(int sz) {
+        return sz >= MIN_BATCH_SIZE && sz <= MAX_BATCH_SIZE;
+    }
+
+    /**
+     * Creates proper batch size error message if {@link #isValidBatchSize(int)} check has failed.
+     *
+     * @param sz The batch size.
+     * @return The string with the error message.
+     */
+    public static String batchSizeErrorMsg(int sz) {
+        return "Batch size should be within [" + MIN_BATCH_SIZE + ".." + MAX_BATCH_SIZE + "]: " + sz;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCacheWriter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCacheWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCacheWriter.java
new file mode 100644
index 0000000..90714c8
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCacheWriter.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteInClosure;
+
+/** A proxy, which stores given key+value pair to a cache. */
+public abstract class BulkLoadCacheWriter implements IgniteInClosure<IgniteBiTuple<?, ?>>, AutoCloseable {
+    /**
+     * Returns number of entry updates made by the writer.
+     *
+     * @return The number of cache entry updates.
+     */
+    public abstract long updateCnt();
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvFormat.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvFormat.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvFormat.java
new file mode 100644
index 0000000..6f5e91e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvFormat.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.regex.Pattern;
+
+/** A placeholder for bulk load CSV format parser options. */
+public class BulkLoadCsvFormat extends BulkLoadFormat {
+
+    /** Line separator pattern. */
+    @NotNull public static final Pattern DEFAULT_LINE_SEPARATOR = Pattern.compile("[\r\n]+");
+
+    /** Field separator pattern. */
+    @NotNull public static final Pattern DEFAULT_FIELD_SEPARATOR = Pattern.compile(",");
+
+    /** Quote characters */
+    @NotNull public static final String DEFAULT_QUOTE_CHARS = "\"";
+
+    /** Default escape sequence start characters. */
+    @Nullable public static final String DEFAULT_ESCAPE_CHARS = null;
+
+    /** Line comment start pattern. */
+    @Nullable public static final Pattern DEFAULT_COMMENT_CHARS = null;
+
+    /** Format name. */
+    public static final String NAME = "CSV";
+
+    /** Line separator pattern. */
+    @Nullable private Pattern lineSeparator;
+
+    /** Field separator pattern. */
+    @Nullable private Pattern fieldSeparator;
+
+    /** Set of quote characters. */
+    @Nullable private String quoteChars;
+
+    /** Line comment start pattern. */
+    @Nullable private Pattern commentChars;
+
+    /** Set of escape start characters. */
+    @Nullable private String escapeChars;
+
+    /**
+     * Returns the name of the format.
+     *
+     * @return The name of the format.
+     */
+    @Override public String name() {
+        return NAME;
+    }
+
+    /**
+     * Returns the line separator pattern.
+     *
+     * @return The line separator pattern.
+     */
+    @Nullable public Pattern lineSeparator() {
+        return lineSeparator;
+    }
+
+    /**
+     * Sets the line separator pattern.
+     *
+     * @param lineSeparator The line separator pattern.
+     */
+    public void lineSeparator(@Nullable Pattern lineSeparator) {
+        this.lineSeparator = lineSeparator;
+    }
+
+    /**
+     * Returns the field separator pattern.
+     *
+     * @return The field separator pattern.
+     */
+    @Nullable public Pattern fieldSeparator() {
+        return fieldSeparator;
+    }
+
+    /**
+     * Sets the field separator pattern.
+     *
+     * @param fieldSeparator The field separator pattern.
+     */
+    public void fieldSeparator(@Nullable Pattern fieldSeparator) {
+        this.fieldSeparator = fieldSeparator;
+    }
+
+    /**
+     * Returns the quote characters.
+     *
+     * @return The quote characters.
+     */
+    @Nullable public String quoteChars() {
+        return quoteChars;
+    }
+
+    /**
+     * Sets the quote characters.
+     *
+     * @param quoteChars The quote characters.
+     */
+    public void quoteChars(@Nullable String quoteChars) {
+        this.quoteChars = quoteChars;
+    }
+
+    /**
+     * Returns the line comment start pattern.
+     *
+     * @return The line comment start pattern.
+     */
+    @Nullable public Pattern commentChars() {
+        return commentChars;
+    }
+
+    /**
+     * Sets the line comment start pattern.
+     *
+     * @param commentChars The line comment start pattern.
+     */
+    public void commentChars(@Nullable Pattern commentChars) {
+        this.commentChars = commentChars;
+    }
+
+    /**
+     * Returns the escape characters.
+     *
+     * @return The escape characters.
+     */
+    @Nullable public String escapeChars() {
+        return escapeChars;
+    }
+
+    /**
+     * Sets the escape characters.
+     *
+     * @param escapeChars The escape characters.
+     */
+    public void escapeChars(@Nullable String escapeChars) {
+        this.escapeChars = escapeChars;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvParser.java
new file mode 100644
index 0000000..0511596
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadCsvParser.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.bulkload.pipeline.CharsetDecoderBlock;
+import org.apache.ignite.internal.processors.bulkload.pipeline.CsvLineProcessorBlock;
+import org.apache.ignite.internal.processors.bulkload.pipeline.PipelineBlock;
+import org.apache.ignite.internal.processors.bulkload.pipeline.StrListAppenderBlock;
+import org.apache.ignite.internal.processors.bulkload.pipeline.LineSplitterBlock;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/** CSV parser for COPY command. */
+public class BulkLoadCsvParser extends BulkLoadParser {
+    /** Processing pipeline input block: a decoder for the input stream of bytes */
+    private final PipelineBlock<byte[], char[]> inputBlock;
+
+    /** A record collecting block that appends its input to {@code List<String>}. */
+    private final StrListAppenderBlock collectorBlock;
+
+    /**
+     * Creates bulk load CSV parser.
+     *
+     *  @param format Format options (parsed from COPY command).
+     */
+    public BulkLoadCsvParser(BulkLoadCsvFormat format) {
+        inputBlock = new CharsetDecoderBlock(BulkLoadFormat.DEFAULT_INPUT_CHARSET);
+
+        collectorBlock = new StrListAppenderBlock();
+
+        // Handling of the other options is to be implemented in IGNITE-7537.
+        inputBlock.append(new LineSplitterBlock(format.lineSeparator()))
+               .append(new CsvLineProcessorBlock(format.fieldSeparator(), format.quoteChars()))
+               .append(collectorBlock);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Iterable<List<Object>> parseBatch(byte[] batchData, boolean isLastBatch)
+        throws IgniteCheckedException {
+        List<List<Object>> res = new LinkedList<>();
+
+        collectorBlock.output(res);
+
+        inputBlock.accept(batchData, isLastBatch);
+
+        return res;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadFormat.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadFormat.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadFormat.java
new file mode 100644
index 0000000..cff93c5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadFormat.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import java.nio.charset.Charset;
+
+/** A superclass and a factory for bulk load format options. */
+public abstract class BulkLoadFormat {
+    /** The default input charset. */
+    public static final Charset DEFAULT_INPUT_CHARSET = Charset.forName("UTF-8");
+
+    /**
+     * Returns the format name.
+     *
+     * @return The format name.
+     */
+    public abstract String name();
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadParser.java
new file mode 100644
index 0000000..252e87b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadParser.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.apache.ignite.IgniteCheckedException;
+
+import java.util.List;
+
+/**
+ * Bulk load file format parser superclass + factory of known formats.
+ *
+ * <p>The parser processes a batch of input data and return a list of records.
+ *
+ * <p>The parser uses corresponding options from {@link BulkLoadFormat} subclass.
+ */
+public abstract class BulkLoadParser {
+    /**
+     * Parses a batch of input data and returns a list of records parsed
+     * (in most cases this is a list of strings).
+     *
+     * <p>Note that conversion between parsed and database table type is done by the other
+     * object (see {@link BulkLoadProcessor#dataConverter}) by the request processing code.
+     * This method is not obliged to do this conversion.
+     *
+     * @param batchData Data from the current batch.
+     * @param isLastBatch true if this is the last batch.
+     * @return The list of records.
+     * @throws IgniteCheckedException If any processing error occurs.
+     */
+    protected abstract Iterable<List<Object>> parseBatch(byte[] batchData, boolean isLastBatch)
+        throws IgniteCheckedException;
+
+    /**
+     * Creates a parser for a given format options.
+     *
+     * @param format The input format object.
+     * @return The parser.
+     * @throws IllegalArgumentException if the format is not known to the factory.
+     */
+    public static BulkLoadParser createParser(BulkLoadFormat format) {
+        if (format instanceof BulkLoadCsvFormat)
+            return new BulkLoadCsvParser((BulkLoadCsvFormat)format);
+
+        throw new IllegalArgumentException("Internal error: format is not defined");
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/25d38cc9/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
new file mode 100644
index 0000000..ccf3e25
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.bulkload;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteIllegalStateException;
+import org.apache.ignite.internal.util.lang.IgniteClosureX;
+import org.apache.ignite.lang.IgniteBiTuple;
+
+import java.util.List;
+
+/**
+ * Bulk load (COPY) command processor used on server to keep various context data and process portions of input
+ * received from the client side.
+ */
+public class BulkLoadProcessor implements AutoCloseable {
+    /** Parser of the input bytes. */
+    private final BulkLoadParser inputParser;
+
+    /**
+     * Converter, which transforms the list of strings parsed from the input stream to the key+value entry to add to
+     * the cache.
+     */
+    private final IgniteClosureX<List<?>, IgniteBiTuple<?, ?>> dataConverter;
+
+    /** Streamer that puts actual key/value into the cache. */
+    private final BulkLoadCacheWriter outputStreamer;
+
+    /** Becomes true after {@link #close()} method is called. */
+    private boolean isClosed;
+
+    /**
+     * Creates bulk load processor.
+     *
+     * @param inputParser Parser of the input bytes.
+     * @param dataConverter Converter, which transforms the list of strings parsed from the input stream to the
+     *     key+value entry to add to the cache.
+     * @param outputStreamer Streamer that puts actual key/value into the cache.
+     */
+    public BulkLoadProcessor(BulkLoadParser inputParser, IgniteClosureX<List<?>, IgniteBiTuple<?, ?>> dataConverter,
+        BulkLoadCacheWriter outputStreamer) {
+        this.inputParser = inputParser;
+        this.dataConverter = dataConverter;
+        this.outputStreamer = outputStreamer;
+        isClosed = false;
+    }
+
+    /**
+     * Returns the streamer that puts actual key/value into the cache.
+     *
+     * @return Streamer that puts actual key/value into the cache.
+     */
+    public BulkLoadCacheWriter outputStreamer() {
+        return outputStreamer;
+    }
+
+    /**
+     * Processes the incoming batch and writes data to the cache by calling the data converter and output streamer.
+     *
+     * @param batchData Data from the current batch.
+     * @param isLastBatch true if this is the last batch.
+     * @throws IgniteIllegalStateException when called after {@link #close()}.
+     */
+    public void processBatch(byte[] batchData, boolean isLastBatch) throws IgniteCheckedException {
+        if (isClosed)
+            throw new IgniteIllegalStateException("Attempt to process a batch on a closed BulkLoadProcessor");
+
+        Iterable<List<Object>> inputRecords = inputParser.parseBatch(batchData, isLastBatch);
+
+        for (List<Object> record : inputRecords) {
+            IgniteBiTuple<?, ?> kv = dataConverter.apply(record);
+
+            outputStreamer.apply(kv);
+        }
+    }
+
+    /**
+     * Aborts processing and closes the underlying objects ({@link IgniteDataStreamer}).
+     */
+    @Override public void close() throws Exception {
+        if (isClosed)
+            return;
+
+        isClosed = true;
+
+        outputStreamer.close();
+    }
+}


Mime
View raw message