diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/CompiledStatements.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/CompiledStatements.java
index 01b8527..a39bc06 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/CompiledStatements.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/CompiledStatements.java
@@ -33,6 +33,7 @@
 import org.apache.asterix.metadata.entities.Dataset;
 import org.apache.asterix.metadata.entities.Datatype;
 import org.apache.asterix.metadata.entities.Index;
+import org.apache.asterix.om.types.ARecordType;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.api.exceptions.SourceLocation;
 
@@ -605,6 +606,8 @@
         private final List<Expression> keyExpressions;
         private final boolean autogenerated;
 
+        private final ARecordType itemType;
+
         public CompiledCopyToStatement(CopyToStatement copyToStatement) {
             this.query = copyToStatement.getQuery();
             this.sourceVariable = copyToStatement.getSourceVariable();
@@ -619,6 +622,7 @@
             this.orderByNullModifierList = copyToStatement.getOrderByNullModifierList();
             this.keyExpressions = copyToStatement.getKeyExpressions();
             this.autogenerated = copyToStatement.isAutogenerated();
+            this.itemType = eddDecl.getItemType();
         }
 
         @Override
@@ -642,6 +646,10 @@
             return properties;
         }
 
+        public ARecordType getItemType() {
+            return itemType;
+        }
+
         public List<Expression> getPathExpressions() {
             return pathExpressions;
         }
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
index d5c916f..978997c 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
@@ -466,7 +466,8 @@
         }
 
         // Write adapter configuration
-        WriteDataSink writeDataSink = new WriteDataSink(copyTo.getAdapter(), copyTo.getProperties());
+        WriteDataSink writeDataSink = new WriteDataSink(copyTo.getAdapter(), copyTo.getProperties(),
+                copyTo.getItemType(), expr.getSourceLocation());
 
         // writeOperator
         WriteOperator writeOperator = new WriteOperator(sourceExprRef, new MutableObject<>(fullPathExpr),
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index d1d61a0..555b6b9 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -27,6 +27,7 @@
 import java.io.InputStream;
 import java.rmi.RemoteException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
@@ -216,6 +217,7 @@
 import org.apache.asterix.om.base.IAObject;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.AUnionType;
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.om.types.BuiltinTypeMap;
 import org.apache.asterix.om.types.IAType;
@@ -4131,6 +4133,25 @@
                             SchemaConverterVisitor.convertToParquetSchemaString((ARecordType) iaType));
                 }
 
+                if (edd.getProperties().get(ExternalDataConstants.KEY_FORMAT)
+                        .equalsIgnoreCase(ExternalDataConstants.FORMAT_CSV_LOWER_CASE)) {
+                    DataverseName dataverseName =
+                            DataverseName.createFromCanonicalForm(ExternalDataConstants.DUMMY_DATAVERSE_NAME);
+                    IAType iaType;
+                    if (copyTo.getType() != null) {
+                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dataverseName,
+                                ExternalDataConstants.DUMMY_TYPE_NAME, copyTo.getType(), mdTxnCtx);
+                    } else if (copyTo.getTypeExpressionItemType() != null) {
+                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dataverseName,
+                                ExternalDataConstants.DUMMY_TYPE_NAME, copyTo.getTypeExpressionItemType(), mdTxnCtx);
+                    } else {
+                        throw new CompilationException(ErrorCode.COMPILATION_ERROR,
+                                "TYPE/AS Expression is required for csv format");
+                    }
+                    ARecordType recordType = (ARecordType) iaType;
+                    validateCSVSchema(recordType);
+                    edd.setItemType(recordType);
+                }
                 Map<VarIdentifier, IAObject> externalVars = createExternalVariables(copyTo, stmtParams);
                 // Query Rewriting (happens under the same ongoing metadata transaction)
                 LangRewritingContext langRewritingContext = createLangRewritingContext(metadataProvider,
@@ -5854,4 +5875,24 @@
         REPLACED
     }
 
+    private void validateCSVSchema(ARecordType schema) throws CompilationException {
+        final List<String> expectedFieldNames = Arrays.asList(schema.getFieldNames());
+        final List<IAType> expectedFieldTypes = Arrays.asList(schema.getFieldTypes());
+        final int size = expectedFieldNames.size();
+        for (int i = 0; i < size; ++i) {
+            IAType expectedIAType = expectedFieldTypes.get(i);
+            if (!ExternalDataConstants.CSV_WRITER_SUPPORTED_DATA_TYPES.contains(expectedIAType.getTypeTag())) {
+                if (expectedIAType.getTypeTag().equals(ATypeTag.UNION)) {
+                    AUnionType unionType = (AUnionType) expectedIAType;
+                    ATypeTag actualTypeTag = unionType.getActualType().getTypeTag();
+                    if (!ExternalDataConstants.CSV_WRITER_SUPPORTED_DATA_TYPES.contains(actualTypeTag)) {
+                        throw new CompilationException(ErrorCode.TYPE_UNSUPPORTED_CSV_WRITE, actualTypeTag.toString());
+                    }
+                } else {
+                    throw new CompilationException(ErrorCode.TYPE_UNSUPPORTED_CSV_WRITE,
+                            expectedIAType.getTypeTag().toString());
+                }
+            }
+        }
+    }
 }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.01.ddl.sqlpp
new file mode 100644
index 0000000..80a8c34
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.01.ddl.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE ColumnType AS {
+  id: bigint,
+  name: string,
+  amount: float,
+  accountNumber: double
+};
+
+CREATE COLLECTION TestCollection(ColumnType) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.02.update.sqlpp
new file mode 100644
index 0000000..d57a311
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.02.update.sqlpp
@@ -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.
+ */
+USE test;
+
+INSERT INTO TestCollection({"id":1, "name":"Macbook1", "amount":123.2, "accountNumber":345.34});
+INSERT INTO TestCollection({"id":2, "name":"Macbook2", "amount":456.7, "accountNumber":123.45});
+INSERT INTO TestCollection({"id":3, "name":"Macbook3", "amount":789.1, "accountNumber":678.90});
+INSERT INTO TestCollection({"id":4, "name":"Macbook4", "amount":234.5, "accountNumber":567.89});
+INSERT INTO TestCollection({"id":5, "name":"Macbook5", "amount":876.5, "accountNumber":345.67});
+INSERT INTO TestCollection({"id":6, "name":"Macbook6", "amount":345.6, "accountNumber":987.65});
+INSERT INTO TestCollection({"id":7, "name":"Macbook7", "amount":678.9, "accountNumber":234.56});
+INSERT INTO TestCollection({"id":8, "name":"Macbook8", "amount":987.2, "accountNumber":789.12});
+INSERT INTO TestCollection({"id":9, "name":"Macbook9", "amount":543.2, "accountNumber":321.45});
+INSERT INTO TestCollection({"id":10, "name":"Macbook10", "amount":123.9, "accountNumber":654.32});
+INSERT INTO TestCollection({"id":11, "name":"Macbook11", "amount":567.8, "accountNumber":456.78});
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.03.update.sqlpp
new file mode 100644
index 0000000..c04f2fe
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.03.update.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+USE test;
+
+COPY (
+   SELECT id, name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "delimiter")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.04.ddl.sqlpp
new file mode 100644
index 0000000..b19cb4b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.04.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="false"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/delimiter"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.05.query.sqlpp
new file mode 100644
index 0000000..b407ec4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.05.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.06.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.06.ddl.sqlpp
new file mode 100644
index 0000000..05d7b83
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.06.ddl.sqlpp
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+/*
+ * Wrong delimiter passed so expected nothing in the output
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopyWrong(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="false"),
+("delimiter"=","),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/delimiter"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.07.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.07.query.sqlpp
new file mode 100644
index 0000000..7e6f1c5
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/delimiter/delimiter.07.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopyWrong d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.01.ddl.sqlpp
new file mode 100644
index 0000000..80a8c34
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.01.ddl.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE ColumnType AS {
+  id: bigint,
+  name: string,
+  amount: float,
+  accountNumber: double
+};
+
+CREATE COLLECTION TestCollection(ColumnType) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.02.update.sqlpp
new file mode 100644
index 0000000..d57a311
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.02.update.sqlpp
@@ -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.
+ */
+USE test;
+
+INSERT INTO TestCollection({"id":1, "name":"Macbook1", "amount":123.2, "accountNumber":345.34});
+INSERT INTO TestCollection({"id":2, "name":"Macbook2", "amount":456.7, "accountNumber":123.45});
+INSERT INTO TestCollection({"id":3, "name":"Macbook3", "amount":789.1, "accountNumber":678.90});
+INSERT INTO TestCollection({"id":4, "name":"Macbook4", "amount":234.5, "accountNumber":567.89});
+INSERT INTO TestCollection({"id":5, "name":"Macbook5", "amount":876.5, "accountNumber":345.67});
+INSERT INTO TestCollection({"id":6, "name":"Macbook6", "amount":345.6, "accountNumber":987.65});
+INSERT INTO TestCollection({"id":7, "name":"Macbook7", "amount":678.9, "accountNumber":234.56});
+INSERT INTO TestCollection({"id":8, "name":"Macbook8", "amount":987.2, "accountNumber":789.12});
+INSERT INTO TestCollection({"id":9, "name":"Macbook9", "amount":543.2, "accountNumber":321.45});
+INSERT INTO TestCollection({"id":10, "name":"Macbook10", "amount":123.9, "accountNumber":654.32});
+INSERT INTO TestCollection({"id":11, "name":"Macbook11", "amount":567.8, "accountNumber":456.78});
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.03.update.sqlpp
new file mode 100644
index 0000000..cc0c717
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.03.update.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+USE test;
+
+COPY (
+   SELECT id, name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "header")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.04.ddl.sqlpp
new file mode 100644
index 0000000..f65d0d4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.04.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/header"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.05.query.sqlpp
new file mode 100644
index 0000000..b407ec4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.05.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.10.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.10.update.sqlpp
new file mode 100644
index 0000000..3b13ef0
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.10.update.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+/*
+ * Default Null Test
+ */
+
+USE test;
+
+COPY (
+   SELECT id, null name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "default", "null")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.11.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.11.ddl.sqlpp
new file mode 100644
index 0000000..1d6cd22
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.11.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopyDefaultNull(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/default/null"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.12.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.12.query.sqlpp
new file mode 100644
index 0000000..bff57e4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.12.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopyDefaultNull dn order by dn.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.20.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.20.update.sqlpp
new file mode 100644
index 0000000..46ee936
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.20.update.sqlpp
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   SELECT id, null name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "custom", "null")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true",
+    "null":"IamNull"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.21.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.21.ddl.sqlpp
new file mode 100644
index 0000000..82f048d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.21.ddl.sqlpp
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+/*
+ * Custom Null Test
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopyCustomNull(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/custom/null"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.22.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.22.query.sqlpp
new file mode 100644
index 0000000..72f9ff3
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.22.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopyCustomNull dn order by dn.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.30.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.30.update.sqlpp
new file mode 100644
index 0000000..1fcf3cd
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.30.update.sqlpp
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   SELECT id, null name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "notUnknown")
+AS (id bigint, name STRING not unknown, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true",
+    "null":"IamNull"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.31.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.31.ddl.sqlpp
new file mode 100644
index 0000000..eb00e4d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.31.ddl.sqlpp
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+/*
+ * Custom Null Test
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopyNotUnknown(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/notUnknown"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.32.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.32.query.sqlpp
new file mode 100644
index 0000000..a6f1945
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/header/header.32.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopyNotUnknown dn order by dn.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.01.ddl.sqlpp
new file mode 100644
index 0000000..80a8c34
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.01.ddl.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE ColumnType AS {
+  id: bigint,
+  name: string,
+  amount: float,
+  accountNumber: double
+};
+
+CREATE COLLECTION TestCollection(ColumnType) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.02.update.sqlpp
new file mode 100644
index 0000000..d57a311
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.02.update.sqlpp
@@ -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.
+ */
+USE test;
+
+INSERT INTO TestCollection({"id":1, "name":"Macbook1", "amount":123.2, "accountNumber":345.34});
+INSERT INTO TestCollection({"id":2, "name":"Macbook2", "amount":456.7, "accountNumber":123.45});
+INSERT INTO TestCollection({"id":3, "name":"Macbook3", "amount":789.1, "accountNumber":678.90});
+INSERT INTO TestCollection({"id":4, "name":"Macbook4", "amount":234.5, "accountNumber":567.89});
+INSERT INTO TestCollection({"id":5, "name":"Macbook5", "amount":876.5, "accountNumber":345.67});
+INSERT INTO TestCollection({"id":6, "name":"Macbook6", "amount":345.6, "accountNumber":987.65});
+INSERT INTO TestCollection({"id":7, "name":"Macbook7", "amount":678.9, "accountNumber":234.56});
+INSERT INTO TestCollection({"id":8, "name":"Macbook8", "amount":987.2, "accountNumber":789.12});
+INSERT INTO TestCollection({"id":9, "name":"Macbook9", "amount":543.2, "accountNumber":321.45});
+INSERT INTO TestCollection({"id":10, "name":"Macbook10", "amount":123.9, "accountNumber":654.32});
+INSERT INTO TestCollection({"id":11, "name":"Macbook11", "amount":567.8, "accountNumber":456.78});
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.03.update.sqlpp
new file mode 100644
index 0000000..b101ae5
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.03.update.sqlpp
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+USE test;
+
+COPY (
+   SELECT id, null name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "null")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true",
+    "null":"IamNull"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.04.ddl.sqlpp
new file mode 100644
index 0000000..f3fdc90
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.04.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopyNull(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/null"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.05.query.sqlpp
new file mode 100644
index 0000000..28beb5d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/null/null.05.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopyNull d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.01.ddl.sqlpp
new file mode 100644
index 0000000..9ee53a6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.01.ddl.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+/* Currently quote and escape are not supported while creating external dataset
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE ColumnType AS {
+  id: bigint,
+  name: string,
+  amount: float,
+  accountNumber: double
+};
+
+CREATE COLLECTION TestCollection(ColumnType) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.02.update.sqlpp
new file mode 100644
index 0000000..8dd62e7
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.02.update.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+USE test;
+
+INSERT INTO TestCollection({"id":1, "name":"Macbook1", "amount":123.2, "accountNumber":345.34});
+INSERT INTO TestCollection({"id":2, "name":"Macbook2", "amount":456.7, "accountNumber":123.45});
+INSERT INTO TestCollection({"id":3, "name":"Macbook3", "amount":789.1, "accountNumber":678.90});
+INSERT INTO TestCollection({"id":4, "name":"Mac|,book4", "amount":234.5, "accountNumber":567.89});
+
+COPY (
+   SELECT id, name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "escape")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "header":"true",
+    "escape":"|"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.03.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.03.ddl.sqlpp
new file mode 100644
index 0000000..2ac9258
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.03.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("escape"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/escape"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.04.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.04.query.sqlpp
new file mode 100644
index 0000000..b407ec4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.04.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.11.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.11.update.sqlpp
new file mode 100644
index 0000000..b468831
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.11.update.sqlpp
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+USE test;
+
+COPY (
+   SELECT id, name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "escape", "1")
+AS (id bigint, name STRING, amount float, accountNumber double)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "header":"true",
+    "escape":"|",
+    "quote": "NONE"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.12.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.12.ddl.sqlpp
new file mode 100644
index 0000000..9224e3c
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.12.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy1(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("escape"="|"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/escape/1"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.13.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.13.query.sqlpp
new file mode 100644
index 0000000..cc786e9
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/quote-escape/quote-escape.13.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy1 d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.01.ddl.sqlpp
new file mode 100644
index 0000000..1e43374
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.01.ddl.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE ColumnType AS {
+  id: bigint,
+  name: string?,
+  amount: float,
+  accountNumber: double
+};
+
+CREATE COLLECTION TestCollection(ColumnType) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.02.update.sqlpp
new file mode 100644
index 0000000..80b535c
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.02.update.sqlpp
@@ -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.
+ */
+USE test;
+
+INSERT INTO TestCollection({"id":1, "name":"", "amount":123.2, "accountNumber":345.34});
+INSERT INTO TestCollection({"id":2, "name":"Macbook2", "amount":456.7, "accountNumber":123.45});
+INSERT INTO TestCollection({"id":3, "name":"Macbook3", "amount":789.1, "accountNumber":678.90});
+INSERT INTO TestCollection({"id":4, "name":"Macbook4", "amount":234.5, "accountNumber":567.89});
+INSERT INTO TestCollection({"id":5, "name":"Macbook5", "amount":876.5, "accountNumber":345.67});
+INSERT INTO TestCollection({"id":6, "name":"Macbook6", "amount":345.6, "accountNumber":987.65});
+INSERT INTO TestCollection({"id":7, "name":"Macbook7", "amount":678.9, "accountNumber":234.56});
+INSERT INTO TestCollection({"id":8, "name":"Macbook8", "amount":987.2, "accountNumber":789.12});
+INSERT INTO TestCollection({"id":9, "name":"Macbook9", "amount":543.2, "accountNumber":321.45});
+INSERT INTO TestCollection({"id":10, "name":"Macbook10", "amount":123.9, "accountNumber":654.32});
+INSERT INTO TestCollection({"id":11, "name":"Macbook11", "amount":567.8, "accountNumber":456.78});
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.03.update.sqlpp
new file mode 100644
index 0000000..cddab9e
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.03.update.sqlpp
@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+/*
+ * This test handles the case with all the CSV knobs. Schema type is JSON style.
+ */
+
+USE test;
+
+COPY (
+   SELECT id, null name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "simple-csv", "1")
+TYPE ( {id: bigint, name: string?, amount: float, accountNumber: double} )
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true",
+    "null":"IamNull",
+    "quote":"'",
+    "force-quote":"false",
+    "escape":"\\",
+    "empty_field_as_null":"true"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.04.ddl.sqlpp
new file mode 100644
index 0000000..c38e775d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.04.ddl.sqlpp
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy1(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("quote"="'"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/simple-csv/1"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.05.query.sqlpp
new file mode 100644
index 0000000..cc786e9
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.05.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy1 d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.11.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.11.update.sqlpp
new file mode 100644
index 0000000..5a8bd3a
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.11.update.sqlpp
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+
+/*
+ * This test handles the case with all the CSV knobs. Schema type is JSON style.
+ * Check for the case when putting missing as null values.
+ */
+
+USE test;
+
+COPY (
+   SELECT id, name, amount, accountNumber FROM TestCollection
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "simple-csv", "2")
+TYPE ( {id: bigint, name: string?, amount: float, accountNumber: double} )
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter":"|",
+    "header":"true",
+    "null":"IamNull",
+    "quote":"'",
+    "force-quote":"false",
+    "escape":"\\",
+    "empty_field_as_null":"true"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.12.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.12.ddl.sqlpp
new file mode 100644
index 0000000..07577d0
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.12.ddl.sqlpp
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy2(ColumnType) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="true"),
+("delimiter"="|"),
+("quote"="'"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/simple-csv/2"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.13.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.13.query.sqlpp
new file mode 100644
index 0000000..547b512
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/simple-csv/simple-csv.13.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id, name, amount, accountNumber
+FROM DatasetCopy2 d order by d.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.01.ddl.sqlpp
new file mode 100644
index 0000000..6228097
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.01.ddl.sqlpp
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.02.update.sqlpp
new file mode 100644
index 0000000..ac5b0b9
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.02.update.sqlpp
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+USE test;
+
+COPY (
+   SELECT "123" as id
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv", "type-mismatch")
+AS (id bigint)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv"
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.03.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.03.ddl.sqlpp
new file mode 100644
index 0000000..8daf039
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.03.ddl.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE EXTERNAL DATASET DatasetCopy(id String) USING S3
+(
+("accessKeyId"="dummyAccessKey"),
+("secretAccessKey"="dummySecretKey"),
+("sessionToken"="dummySessionToken"),
+("header"="false"),
+("region"="us-west-2"),
+  ("serviceEndpoint"="http://127.0.0.1:8001"),
+  ("container"="playground"),
+  ("definition"="copy-to-result/csv/type-mismatch"),
+  ("format" = "csv"),
+  ("requireVersionChangeDetection"="false"),
+  ("include"="*.csv")
+);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.04.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.04.query.sqlpp
new file mode 100644
index 0000000..8e2b639
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/csv/type-mismatch/type-mismatch.04.query.sqlpp
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+
+SELECT id
+FROM DatasetCopy c order by c.id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.01.ddl.sqlpp
new file mode 100644
index 0000000..591c949
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.01.ddl.sqlpp
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test if exists;
+CREATE DATAVERSE test;
+USE test;
+
+
+CREATE TYPE ColumnType1 AS {
+  id: integer
+};
+
+CREATE COLLECTION TestCollection(ColumnType1) PRIMARY KEY id;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.02.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.02.ddl.sqlpp
new file mode 100644
index 0000000..ef6631b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.02.ddl.sqlpp
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks2")
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.03.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.03.ddl.sqlpp
new file mode 100644
index 0000000..87ebe40
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.03.ddl.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+AS (id wrongDataType)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.04.ddl.sqlpp
new file mode 100644
index 0000000..117fbde
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.04.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+AS (id bigint)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "quote": "ABCD"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.05.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.05.ddl.sqlpp
new file mode 100644
index 0000000..aed1ed4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.05.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+AS (id bigint)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "delimiter": "wrongDelimiter"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.06.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.06.ddl.sqlpp
new file mode 100644
index 0000000..e51644b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.06.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+AS (id bigint)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "escape": "wrongEscape"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.07.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.07.ddl.sqlpp
new file mode 100644
index 0000000..2509d18
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.07.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+AS (id bigint)
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "record-delimiter": "ABCD"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.08.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.08.ddl.sqlpp
new file mode 100644
index 0000000..a995d7a
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.08.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+TYPE ( { id : int,  name : { first :  [ string ] } } )
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "record-delimiter": ","
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.09.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.09.ddl.sqlpp
new file mode 100644
index 0000000..e2d7b4f
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.09.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+TYPE ( { id : int,  name : [ string ] } )
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "record-delimiter": ","
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.10.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.10.ddl.sqlpp
new file mode 100644
index 0000000..8706be3
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/csv-error-checks/csv-error-checks.10.ddl.sqlpp
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+   select c.* from TestCollection c
+) toWriter
+TO S3
+PATH ("copy-to-result", "csv-error-checks3")
+TYPE ( { id : int,  name : string } )
+AS ( id string )
+WITH {
+    "accessKeyId":"dummyAccessKey",
+    "secretAccessKey":"dummySecretKey",
+    "region":"us-west-2",
+    "serviceEndpoint":"http://127.0.0.1:8001",
+    "container":"playground",
+    "format":"csv",
+    "record-delimiter": ","
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/supported-adapter-format-compression/supported-adapters.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/supported-adapter-format-compression/supported-adapters.03.update.sqlpp
index 2e227d6..a9fe8cf 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/supported-adapter-format-compression/supported-adapters.03.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/supported-adapter-format-compression/supported-adapters.03.update.sqlpp
@@ -28,7 +28,7 @@
     "region":"us-west-2",
     "serviceEndpoint":"http://127.0.0.1:8001",
     "container":"playground",
-    "format":"csv"
+    "format":"avro"
 }
 
 
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.05.adm
new file mode 100644
index 0000000..8722e43
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.05.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "Macbook1", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "Macbook2", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "Macbook3", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "Macbook4", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "Macbook5", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "Macbook6", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "Macbook7", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "Macbook8", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "Macbook9", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "Macbook10", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "Macbook11", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.07.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.07.adm
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/delimiter/delimiter.07.adm
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.05.adm
new file mode 100644
index 0000000..8722e43
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.05.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "Macbook1", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "Macbook2", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "Macbook3", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "Macbook4", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "Macbook5", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "Macbook6", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "Macbook7", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "Macbook8", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "Macbook9", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "Macbook10", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "Macbook11", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.12.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.12.adm
new file mode 100644
index 0000000..5055d28
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.12.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.22.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.22.adm
new file mode 100644
index 0000000..90e44c4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.22.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "IamNull", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "IamNull", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "IamNull", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "IamNull", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "IamNull", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "IamNull", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "IamNull", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "IamNull", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "IamNull", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "IamNull", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "IamNull", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.32.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.32.adm
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/header/header.32.adm
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/null/null.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/null/null.05.adm
new file mode 100644
index 0000000..90e44c4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/null/null.05.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "IamNull", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "IamNull", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "IamNull", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "IamNull", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "IamNull", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "IamNull", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "IamNull", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "IamNull", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "IamNull", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "IamNull", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "IamNull", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.04.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.04.adm
new file mode 100644
index 0000000..2f9e206
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.04.adm
@@ -0,0 +1,4 @@
+{ "id": 1, "name": "Macbook1", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "Macbook2", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "Macbook3", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "Mac|,book4", "amount": 234.5, "accountNumber": 567.89 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.13.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.13.adm
new file mode 100644
index 0000000..2f9e206
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/quote-escape/quote-escape.13.adm
@@ -0,0 +1,4 @@
+{ "id": 1, "name": "Macbook1", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "Macbook2", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "Macbook3", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "Mac|,book4", "amount": 234.5, "accountNumber": 567.89 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.05.adm
new file mode 100644
index 0000000..90e44c4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.05.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "IamNull", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "IamNull", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "IamNull", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "IamNull", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "IamNull", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "IamNull", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "IamNull", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "IamNull", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "IamNull", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "IamNull", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "IamNull", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.13.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.13.adm
new file mode 100644
index 0000000..2c305f4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/simple-csv/simple-csv.13.adm
@@ -0,0 +1,11 @@
+{ "id": 1, "name": "IamNull", "amount": 123.2, "accountNumber": 345.34 }
+{ "id": 2, "name": "Macbook2", "amount": 456.7, "accountNumber": 123.45 }
+{ "id": 3, "name": "Macbook3", "amount": 789.1, "accountNumber": 678.90 }
+{ "id": 4, "name": "Macbook4", "amount": 234.5, "accountNumber": 567.89 }
+{ "id": 5, "name": "Macbook5", "amount": 876.5, "accountNumber": 345.67 }
+{ "id": 6, "name": "Macbook6", "amount": 345.6, "accountNumber": 987.65 }
+{ "id": 7, "name": "Macbook7", "amount": 678.9, "accountNumber": 234.56 }
+{ "id": 8, "name": "Macbook8", "amount": 987.2, "accountNumber": 789.12 }
+{ "id": 9, "name": "Macbook9", "amount": 543.2, "accountNumber": 321.45 }
+{ "id": 10, "name": "Macbook10", "amount": 123.9, "accountNumber": 654.32 }
+{ "id": 11, "name": "Macbook11", "amount": 567.8, "accountNumber": 456.78 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/type-mismatch/type-mismatch.04.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/type-mismatch/type-mismatch.04.adm
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/csv/type-mismatch/type-mismatch.04.adm
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
index aadc42b..80628ab 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
@@ -105,7 +105,7 @@
       <compilation-unit name="supported-adapter-format-compression">
         <output-dir compare="Text">supported-adapter-format-compression</output-dir>
         <expected-error>ASX1188: Unsupported writing adapter 'AZUREBLOB'. Supported adapters: [gcs, localfs, s3]</expected-error>
-        <expected-error>ASX1189: Unsupported writing format 'csv'. Supported formats: [json, parquet]</expected-error>
+        <expected-error>ASX1189: Unsupported writing format 'avro'. Supported formats: [csv, json, parquet]</expected-error>
         <expected-error>ASX1202: Unsupported compression scheme rar. Supported schemes for json are [gzip]</expected-error>
       </compilation-unit>
     </test-case>
@@ -141,6 +141,47 @@
         <expected-error>Expected integer value, got hello</expected-error>
       </compilation-unit>
     </test-case>
+    <test-case FilePath="copy-to/negative">
+      <compilation-unit name="csv-error-checks">
+        <output-dir compare="Text">csv-error-checks</output-dir>
+        <expected-error>ASX1079: Compilation error: TYPE/AS Expression is required for csv format</expected-error>
+        <expected-error>ASX1082: Cannot find datatype with name wrongDataType (in line 27, at column 4)</expected-error>
+        <expected-error>ASX3124: 'ABCD' is not a valid quote. The length of a quote should be 1</expected-error>
+        <expected-error>ASX3049: 'wrongDelimiter' is not a valid delimiter. The length of a delimiter should be 1</expected-error>
+        <expected-error>ASX3126: 'wrongEscape' is not a valid escape. The length of a escape should be 1</expected-error>
+        <expected-error>ASX3125: 'ABCD' is not a valid force-quote input. The length of a force-quote input should be 1 character</expected-error>
+        <expected-error>ASX1207: 'object' type not supported in csv format</expected-error>
+        <expected-error>ASX1207: 'array' type not supported in csv format</expected-error>
+        <expected-error>Syntax error: Both 'TYPE()' and 'AS()' are provided. Please use either 'TYPE()' or 'AS()'.</expected-error>
+      </compilation-unit>
+    </test-case>
+  </test-group>
+  <test-group name="copy-to/csv">
+    <test-case FilePath="copy-to/csv">
+      <compilation-unit name="simple-csv">
+        <output-dir compare="Text">simple-csv</output-dir>
+      </compilation-unit>
+    </test-case>
+    <test-case FilePath="copy-to/csv">
+      <compilation-unit name="type-mismatch">
+        <output-dir compare="Text">type-mismatch</output-dir>
+      </compilation-unit>
+    </test-case>
+    <test-case FilePath="copy-to/csv">
+      <compilation-unit name="delimiter">
+        <output-dir compare="Text">delimiter</output-dir>
+      </compilation-unit>
+    </test-case>
+    <test-case FilePath="copy-to/csv">
+      <compilation-unit name="header">
+        <output-dir compare="Text">header</output-dir>
+      </compilation-unit>
+    </test-case>
+    <test-case FilePath="copy-to/csv">
+      <compilation-unit name="quote-escape">
+        <output-dir compare="Text">quote-escape</output-dir>
+      </compilation-unit>
+    </test-case>
   </test-group>
   <test-group name="aws-s3-external-dataset">
     <test-case FilePath="external-dataset">
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index ea7cc3b..550015d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -309,6 +309,8 @@
     TYPE_UNSUPPORTED_PARQUET_WRITE(1204),
     INVALID_PARQUET_WRITER_VERSION(1205),
     ILLEGAL_SIZE_PROVIDED(1206),
+    TYPE_UNSUPPORTED_CSV_WRITE(1207),
+    INVALID_CSV_SCHEMA(1208),
     // Feed errors
     DATAFLOW_ILLEGAL_STATE(3001),
     UTIL_DATAFLOW_UTILS_TUPLE_TOO_LARGE(3002),
@@ -427,6 +429,10 @@
     PARAM_NOT_ALLOWED_IF_PARAM_IS_PRESENT(3122),
     // Avro error
     UNSUPPORTED_TYPE_FOR_AVRO(3123),
+    // Copy to CSV Error
+    INVALID_QUOTE(3124),
+    INVALID_FORCE_QUOTE(3125),
+    INVALID_ESCAPE(3126),
 
     // Lifecycle management errors
     DUPLICATE_PARTITION_ID(4000),
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index 3e2f124..2350c2c 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -311,6 +311,8 @@
 1204 = '%1$s' type not supported in parquet format
 1205 = Invalid Parquet Writer Version provided '%1$s'. Supported values: %2$s
 1206 = Storage units expected for the field '%1$s' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided '%2$s'
+1207 = '%1$s' type not supported in csv format
+1208 = Invalid Copy to CSV schema
 
 # Feed Errors
 3001 = Illegal state.
@@ -432,6 +434,9 @@
 3121 = Parameter '%1$s' or '%2$s' is required if '%3$s' is provided
 3122 = Parameter '%1$s' is not allowed if '%2$s' is provided
 3123 = Type '%1$s' contains declared fields, which is not supported for 'avro' format
+3124 = '%1$s' is not a valid quote. The length of a quote should be 1
+3125 = '%1$s' is not a valid force-quote input. The length of a force-quote input should be 1 character
+3126 = '%1$s' is not a valid escape. The length of a escape should be 1
 
 # Lifecycle management errors
 4000 = Partition id %1$s for node %2$s already in use by node %3$s
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 23aa5dc..18d5158 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -18,6 +18,7 @@
  */
 package org.apache.asterix.external.util;
 
+import java.util.List;
 import java.util.Set;
 import java.util.TimeZone;
 import java.util.function.LongSupplier;
@@ -90,6 +91,9 @@
     public static final String KEY_INCLUDE = "include";
     public static final String KEY_EXCLUDE = "exclude";
     public static final String KEY_QUOTE = "quote";
+    public static final String KEY_FORCE_QUOTE = "force-quote";
+    public static final String KEY_EMPTY_FIELD_AS_NULL = "empty_field_as_null";
+    public static final String KEY_RECORD_DELIMITER = "record-delimiter";
     public static final String KEY_ESCAPE = "escape";
     public static final String KEY_PARSER = "parser";
     public static final String KEY_DATASET_RECORD = "dataset-record";
@@ -188,6 +192,8 @@
     public static final String HAS_HEADER = "has.header";
     public static final String TIME_TRACKING = "time.tracking";
     public static final String DEFAULT_QUOTE = "\"";
+    public static final String DEFAULT_SINGLE_QUOTE = "'";
+    public static final String NONE = "none";
     public static final String NODE_RESOLVER_FACTORY_PROPERTY = "node.Resolver";
     public static final String DEFAULT_DELIMITER = ",";
     public static final String EXTERNAL_LIBRARY_SEPARATOR = "#";
@@ -199,6 +205,7 @@
     public static final String FORMAT_ADM = "adm";
     public static final String FORMAT_AVRO = "avro";
     public static final String FORMAT_JSON_LOWER_CASE = "json";
+    public static final String FORMAT_CSV_LOWER_CASE = "csv";
     public static final String FORMAT_JSON_UPPER_CASE = "JSON";
     public static final String FORMAT_DELIMITED_TEXT = "delimited-text";
     public static final String FORMAT_TWEET = "twitter-status";
@@ -339,15 +346,21 @@
     public static final Set<String> PARQUET_WRITER_SUPPORTED_COMPRESSION;
     public static final Set<String> PARQUET_WRITER_SUPPORTED_VERSION;
     public static final int PARQUET_DICTIONARY_PAGE_SIZE = 1048576;
+    public static final List<String> WRITER_SUPPORTED_QUOTES;
+    public static final List<ATypeTag> CSV_WRITER_SUPPORTED_DATA_TYPES =
+            List.of(ATypeTag.TINYINT, ATypeTag.SMALLINT, ATypeTag.INTEGER, ATypeTag.BIGINT, ATypeTag.UINT8,
+                    ATypeTag.UINT16, ATypeTag.UINT64, ATypeTag.FLOAT, ATypeTag.DOUBLE, ATypeTag.STRING,
+                    ATypeTag.BOOLEAN, ATypeTag.DATETIME, ATypeTag.UINT32, ATypeTag.DATE, ATypeTag.TIME);
 
     static {
-        WRITER_SUPPORTED_FORMATS = Set.of(FORMAT_JSON_LOWER_CASE, FORMAT_PARQUET);
+        WRITER_SUPPORTED_FORMATS = Set.of(FORMAT_JSON_LOWER_CASE, FORMAT_PARQUET, FORMAT_CSV_LOWER_CASE);
         WRITER_SUPPORTED_ADAPTERS = Set.of(ALIAS_LOCALFS_ADAPTER.toLowerCase(), KEY_ADAPTER_NAME_AWS_S3.toLowerCase(),
                 KEY_ADAPTER_NAME_GCS.toLowerCase());
         TEXTUAL_WRITER_SUPPORTED_COMPRESSION = Set.of(KEY_COMPRESSION_GZIP);
         PARQUET_WRITER_SUPPORTED_COMPRESSION =
                 Set.of(KEY_COMPRESSION_GZIP, KEY_COMPRESSION_SNAPPY, KEY_COMPRESSION_ZSTD);
         PARQUET_WRITER_SUPPORTED_VERSION = Set.of(PARQUET_WRITER_VERSION_VALUE_1, PARQUET_WRITER_VERSION_VALUE_2);
+        WRITER_SUPPORTED_QUOTES = List.of(DEFAULT_QUOTE, DEFAULT_SINGLE_QUOTE, NONE);
     }
 
     public static class ParquetOptions {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
index b4314b9..c362f75 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
@@ -27,6 +27,7 @@
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_INCLUDE;
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PATH;
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_QUOTE;
+import static org.apache.asterix.external.util.ExternalDataConstants.KEY_READER;
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_RECORD_END;
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_RECORD_START;
 import static org.apache.asterix.external.util.azure.blob_storage.AzureUtils.validateAzureBlobProperties;
@@ -437,6 +438,7 @@
     public static void defaultConfiguration(Map<String, String> configuration) {
         String format = configuration.get(ExternalDataConstants.KEY_FORMAT);
         if (format != null) {
+            //todo:utsav
             // default quote, escape character for quote and fields delimiter for csv and tsv format
             if (format.equals(ExternalDataConstants.FORMAT_CSV)) {
                 configuration.putIfAbsent(KEY_DELIMITER, ExternalDataConstants.DEFAULT_DELIMITER);
@@ -610,12 +612,22 @@
     public static void validate(Map<String, String> configuration) throws HyracksDataException {
         String format = configuration.get(ExternalDataConstants.KEY_FORMAT);
         String header = configuration.get(ExternalDataConstants.KEY_HEADER);
+        String forceQuote = configuration.get(ExternalDataConstants.KEY_FORCE_QUOTE);
+        String emptyFieldAsNull = configuration.get(ExternalDataConstants.KEY_EMPTY_FIELD_AS_NULL);
         if (format != null && isHeaderRequiredFor(format) && header == null) {
             throw new RuntimeDataException(ErrorCode.PARAMETERS_REQUIRED, ExternalDataConstants.KEY_HEADER);
         }
         if (header != null && !isBoolean(header)) {
             throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL, ExternalDataConstants.KEY_HEADER, header);
         }
+        if (forceQuote != null && !isBoolean(forceQuote)) {
+            throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL, ExternalDataConstants.KEY_FORCE_QUOTE,
+                    forceQuote);
+        }
+        if (emptyFieldAsNull != null && !isBoolean(emptyFieldAsNull)) {
+            throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL,
+                    ExternalDataConstants.KEY_EMPTY_FIELD_AS_NULL, emptyFieldAsNull);
+        }
         char delimiter = validateGetDelimiter(configuration);
         validateGetQuote(configuration, delimiter);
         validateGetEscape(configuration, format);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
index 5059ec8..3de4067 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
@@ -21,6 +21,7 @@
 import static org.apache.asterix.common.exceptions.ErrorCode.INVALID_REQ_PARAM_VAL;
 import static org.apache.asterix.common.exceptions.ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM;
 import static org.apache.asterix.common.exceptions.ErrorCode.PARAMETERS_REQUIRED;
+import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_CSV;
 import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_JSON_LOWER_CASE;
 import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_PARQUET;
 import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PARQUET_PAGE_SIZE;
@@ -51,6 +52,15 @@
         validateMaxResult(configuration, sourceLocation);
     }
 
+    private static void validateQuote(Map<String, String> configuration, SourceLocation sourceLocation)
+            throws CompilationException {
+        String quote = configuration.get(ExternalDataConstants.KEY_QUOTE);
+        if (quote != null && !ExternalDataConstants.WRITER_SUPPORTED_QUOTES.contains(quote.toLowerCase())) {
+            throw CompilationException.create(ErrorCode.INVALID_QUOTE, sourceLocation, quote,
+                    ExternalDataConstants.WRITER_SUPPORTED_QUOTES.toString());
+        }
+    }
+
     private static void validateAdapter(String adapter, Set<String> supportedAdapters, SourceLocation sourceLocation)
             throws CompilationException {
         checkSupported(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE, adapter, supportedAdapters,
@@ -69,6 +79,9 @@
             case FORMAT_PARQUET:
                 validateParquet(configuration, sourceLocation);
                 break;
+            case FORMAT_CSV:
+                validateCSV(configuration, sourceLocation);
+                break;
         }
     }
 
@@ -115,6 +128,15 @@
         validateTextualCompression(configuration, sourceLocation);
     }
 
+    private static void validateCSV(Map<String, String> configuration, SourceLocation sourceLocation)
+            throws CompilationException {
+        validateTextualCompression(configuration, sourceLocation);
+        validateDelimiter(configuration, sourceLocation);
+        validateRecordDelimiter(configuration, sourceLocation);
+        validateQuote(configuration, sourceLocation);
+        validateEscape(configuration, sourceLocation);
+    }
+
     private static void validateParquetCompression(Map<String, String> configuration, SourceLocation sourceLocation)
             throws CompilationException {
         String compression = configuration.get(ExternalDataConstants.KEY_WRITER_COMPRESSION);
@@ -205,4 +227,31 @@
         }
     }
 
+    private static void validateDelimiter(Map<String, String> configuration, SourceLocation sourceLocation)
+            throws CompilationException {
+        // Will this affect backward compatibility
+        String delimiter = configuration.get(ExternalDataConstants.KEY_DELIMITER);
+        unitByteCondition(delimiter, sourceLocation, ErrorCode.INVALID_DELIMITER);
+    }
+
+    private static void validateEscape(Map<String, String> configuration, SourceLocation sourceLocation)
+            throws CompilationException {
+        // Will this affect backward compatibility?
+        String escape = configuration.get(ExternalDataConstants.KEY_ESCAPE);
+        unitByteCondition(escape, sourceLocation, ErrorCode.INVALID_ESCAPE);
+    }
+
+    private static void validateRecordDelimiter(Map<String, String> configuration, SourceLocation sourceLocation)
+            throws CompilationException {
+        String recordDel = configuration.get(ExternalDataConstants.KEY_RECORD_DELIMITER);
+        unitByteCondition(recordDel, sourceLocation, ErrorCode.INVALID_FORCE_QUOTE);
+    }
+
+    private static void unitByteCondition(String param, SourceLocation sourceLocation, ErrorCode errorCode)
+            throws CompilationException {
+        if (param != null && param.length() > 1 && param.getBytes().length != 1) {
+            throw CompilationException.create(errorCode, sourceLocation, param);
+        }
+    }
+
 }
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinter.java
new file mode 100644
index 0000000..b5299ad
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinter.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.asterix.external.writer.printer;
+
+import org.apache.asterix.external.writer.compressor.IExternalFileCompressStreamFactory;
+import org.apache.hyracks.algebricks.data.IPrinter;
+
+public class CsvExternalFilePrinter extends AbstractTextualExternalPrinter {
+    CsvExternalFilePrinter(IPrinter printer, IExternalFileCompressStreamFactory compressStreamFactory) {
+        super(printer, compressStreamFactory);
+    }
+
+    @Override
+    void afterPrint() {
+    }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinterFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinterFactory.java
new file mode 100644
index 0000000..0ed1498
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/CsvExternalFilePrinterFactory.java
@@ -0,0 +1,42 @@
+/*
+ * 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.asterix.external.writer.printer;
+
+import org.apache.asterix.external.writer.compressor.IExternalFileCompressStreamFactory;
+import org.apache.asterix.runtime.writer.IExternalPrinter;
+import org.apache.asterix.runtime.writer.IExternalPrinterFactory;
+import org.apache.hyracks.algebricks.data.IPrinterFactory;
+
+public class CsvExternalFilePrinterFactory implements IExternalPrinterFactory {
+    private static final long serialVersionUID = 8971234908711234L;
+    protected final IPrinterFactory printerFactory;
+    private final IExternalFileCompressStreamFactory compressStreamFactory;
+
+    public CsvExternalFilePrinterFactory(IPrinterFactory printerFactory,
+            IExternalFileCompressStreamFactory compressStreamFactory) {
+        this.printerFactory = printerFactory;
+        this.compressStreamFactory = compressStreamFactory;
+    }
+
+    @Override
+    public IExternalPrinter createPrinter() {
+        return new CsvExternalFilePrinter(printerFactory.createPrinter(), compressStreamFactory);
+    }
+}
diff --git a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CopyToStatement.java b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CopyToStatement.java
index 5bc4a71..5c89a9f 100644
--- a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CopyToStatement.java
+++ b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CopyToStatement.java
@@ -55,13 +55,14 @@
     private List<Expression> orderByList;
     private int varCounter;
     private RecordTypeDefinition itemType;
+    private TypeExpression typeExpressionItemType;
 
     public CopyToStatement(Namespace namespace, String datasetName, Query query, VariableExpr sourceVariable,
             ExternalDetailsDecl externalDetailsDecl, int varCounter, List<Expression> keyExpressions,
             boolean autogenerated) {
         this(namespace, datasetName, query, sourceVariable, externalDetailsDecl, new ArrayList<>(), new ArrayList<>(),
                 new HashMap<>(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>(), varCounter, keyExpressions,
-                autogenerated, null);
+                autogenerated, null, null);
     }
 
     public CopyToStatement(Namespace namespace, String datasetName, Query query, VariableExpr sourceVariable,
@@ -71,7 +72,7 @@
             List<OrderbyClause.NullOrderModifier> orderByNullModifierList, int varCounter) {
         this(namespace, datasetName, query, sourceVariable, externalDetailsDecl, pathExpressions, partitionExpressions,
                 partitionsVariables, orderbyList, orderByModifiers, orderByNullModifierList, varCounter,
-                new ArrayList<>(), false, null);
+                new ArrayList<>(), false, null, null);
     }
 
     public CopyToStatement(Namespace namespace, String datasetName, Query query, VariableExpr sourceVariable,
@@ -79,10 +80,10 @@
             List<Expression> partitionExpressions, Map<Integer, VariableExpr> partitionsVariables,
             List<Expression> orderbyList, List<OrderbyClause.OrderModifier> orderByModifiers,
             List<OrderbyClause.NullOrderModifier> orderByNullModifierList, int varCounter,
-            RecordTypeDefinition itemType) {
+            TypeExpression typeExpressionItemType, RecordTypeDefinition itemType) {
         this(namespace, datasetName, query, sourceVariable, externalDetailsDecl, pathExpressions, partitionExpressions,
                 partitionsVariables, orderbyList, orderByModifiers, orderByNullModifierList, varCounter,
-                new ArrayList<>(), false, itemType);
+                new ArrayList<>(), false, typeExpressionItemType, itemType);
     }
 
     private CopyToStatement(Namespace namespace, String datasetName, Query query, VariableExpr sourceVariable,
@@ -90,7 +91,8 @@
             List<Expression> partitionExpressions, Map<Integer, VariableExpr> partitionsVariables,
             List<Expression> orderbyList, List<OrderbyClause.OrderModifier> orderByModifiers,
             List<OrderbyClause.NullOrderModifier> orderByNullModifierList, int varCounter,
-            List<Expression> keyExpressions, boolean autogenerated, RecordTypeDefinition itemType) {
+            List<Expression> keyExpressions, boolean autogenerated, TypeExpression typeExpressionItemType,
+            RecordTypeDefinition itemType) {
         this.namespace = namespace;
         this.datasetName = datasetName;
         this.query = query;
@@ -106,6 +108,7 @@
         this.keyExpressions = keyExpressions != null ? keyExpressions : new ArrayList<>();
         this.autogenerated = autogenerated;
         this.itemType = itemType;
+        this.typeExpressionItemType = typeExpressionItemType;
 
         if (pathExpressions.isEmpty()) {
             // Ensure path expressions to have at least an empty string
@@ -211,8 +214,8 @@
         return !orderByList.isEmpty();
     }
 
-    public TypeExpression getItemType() {
-        return itemType;
+    public TypeExpression getTypeExpressionItemType() {
+        return typeExpressionItemType;
     }
 
     @Override
diff --git a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/ExternalDetailsDecl.java b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/ExternalDetailsDecl.java
index db599c4..e1c978a 100644
--- a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/ExternalDetailsDecl.java
+++ b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/ExternalDetailsDecl.java
@@ -20,9 +20,12 @@
 
 import java.util.Map;
 
+import org.apache.asterix.om.types.ARecordType;
+
 public class ExternalDetailsDecl implements IDatasetDetailsDecl {
     private Map<String, String> properties;
     private String adapter;
+    private ARecordType itemType;
 
     public void setAdapter(String adapter) {
         this.adapter = adapter;
@@ -32,6 +35,14 @@
         this.properties = properties;
     }
 
+    public void setItemType(ARecordType itemType) {
+        this.itemType = itemType;
+    }
+
+    public ARecordType getItemType() {
+        return itemType;
+    }
+
     public String getAdapter() {
         return adapter;
     }
diff --git a/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj b/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj
index 1cfa892..2ec743d 100644
--- a/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj
+++ b/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj
@@ -2936,7 +2936,9 @@
   Namespace namespace = nameComponents == null ? null : nameComponents.first;
   String datasetName = nameComponents == null ? null : nameComponents.second.getValue();
   List<Expression> pathExprs;
-  RecordTypeDefinition typeExpr = null;
+  RecordTypeDefinition recordTypeDefinition = null;
+  TypeExpression typeExpr = null;
+  Boolean isRecordTypeDefinition = false;
 
   List<Expression> partitionExprs = new ArrayList<Expression>();
   Map<Integer, VariableExpr> partitionVarExprs = new HashMap<Integer, VariableExpr>();
@@ -2948,7 +2950,21 @@
   <TO> adapterName = AdapterName()
   <PATH> <LEFTPAREN> pathExprs = ExpressionList() <RIGHTPAREN>
   (CopyToOverClause(partitionExprs, partitionVarExprs, orderbyList, orderbyModifierList, orderbyNullModifierList))?
-  (<TYPE>  <LEFTPAREN> typeExpr = RecordTypeDef() <RIGHTPAREN>)?
+  (<TYPE> <LEFTPAREN>
+    {
+      recordTypeDefinition = RecordTypeDef();
+      isRecordTypeDefinition = true;
+    }
+    <RIGHTPAREN>) ?
+  (<AS>
+    {
+      if (isRecordTypeDefinition == false) {
+        typeExpr = DatasetRecordTypeSpecification(false, null);
+      } else {
+        throw new SqlppParseException(getSourceLocation(token), "Syntax error: Both 'TYPE()' and 'AS()' are provided. Please use either 'TYPE()' or 'AS()'.");
+      }
+    }
+  )?
   <WITH> withRecord = RecordConstructor()
     {
        ExternalDetailsDecl edd = new ExternalDetailsDecl();
@@ -2963,7 +2979,7 @@
           usedAlias = new VariableExpr(SqlppVariableUtil.toInternalVariableIdentifier(datasetName));
        }
 
-       CopyToStatement stmt = new CopyToStatement(namespace, datasetName, query, usedAlias, edd, pathExprs, partitionExprs, partitionVarExprs, orderbyList, orderbyModifierList, orderbyNullModifierList, getVarCounter(), typeExpr);
+       CopyToStatement stmt = new CopyToStatement(namespace, datasetName, query, usedAlias, edd, pathExprs, partitionExprs, partitionVarExprs, orderbyList, orderbyModifierList, orderbyNullModifierList, getVarCounter(), typeExpr, recordTypeDefinition);
        return addSourceLocation(stmt, startToken);
     }
 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/IExternalWriteDataSink.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/IExternalWriteDataSink.java
new file mode 100644
index 0000000..64f8d6d
--- /dev/null
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/IExternalWriteDataSink.java
@@ -0,0 +1,30 @@
+/*
+ * 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.asterix.metadata.declared;
+
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.hyracks.algebricks.core.algebra.metadata.IWriteDataSink;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+
+public interface IExternalWriteDataSink extends IWriteDataSink {
+    ARecordType getItemType();
+
+    SourceLocation getSourceLoc();
+}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/WriteDataSink.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/WriteDataSink.java
index 753ac54..d1667bf 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/WriteDataSink.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/WriteDataSink.java
@@ -21,20 +21,39 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.asterix.om.types.ARecordType;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IWriteDataSink;
+import org.apache.hyracks.api.exceptions.SourceLocation;
 
-public class WriteDataSink implements IWriteDataSink {
+public class WriteDataSink implements IExternalWriteDataSink {
     private final String adapterName;
     private final Map<String, String> configuration;
+    private ARecordType itemType;
+    private SourceLocation sourceLoc;
 
-    public WriteDataSink(String adapterName, Map<String, String> configuration) {
+    public WriteDataSink(String adapterName, Map<String, String> configuration, ARecordType itemType,
+            SourceLocation sourceLoc) {
         this.adapterName = adapterName;
         this.configuration = configuration;
+        this.itemType = itemType;
+        this.sourceLoc = sourceLoc;
     }
 
     private WriteDataSink(WriteDataSink writeDataSink) {
         this.adapterName = writeDataSink.getAdapterName();
         this.configuration = new HashMap<>(writeDataSink.configuration);
+        this.itemType = writeDataSink.itemType;
+        this.sourceLoc = writeDataSink.sourceLoc;
+    }
+
+    @Override
+    public ARecordType getItemType() {
+        return itemType;
+    }
+
+    @Override
+    public SourceLocation getSourceLoc() {
+        return sourceLoc;
     }
 
     @Override
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
index ee7b3fc..73bdbb8 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
@@ -26,15 +26,20 @@
 import org.apache.asterix.cloud.writer.S3ExternalFileWriterFactory;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.exceptions.CompilationException;
+import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.asterix.external.util.ExternalDataConstants;
 import org.apache.asterix.external.util.ExternalDataUtils;
 import org.apache.asterix.external.writer.LocalFSExternalFileWriterFactory;
 import org.apache.asterix.external.writer.compressor.GzipExternalFileCompressStreamFactory;
 import org.apache.asterix.external.writer.compressor.IExternalFileCompressStreamFactory;
 import org.apache.asterix.external.writer.compressor.NoOpExternalFileCompressStreamFactory;
+import org.apache.asterix.external.writer.printer.CsvExternalFilePrinterFactory;
 import org.apache.asterix.external.writer.printer.ParquetExternalFilePrinterFactory;
 import org.apache.asterix.external.writer.printer.TextualExternalFilePrinterFactory;
+import org.apache.asterix.formats.nontagged.CSVPrinterFactoryProvider;
 import org.apache.asterix.formats.nontagged.CleanJSONPrinterFactoryProvider;
+import org.apache.asterix.metadata.declared.IExternalWriteDataSink;
+import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.runtime.writer.ExternalFileWriterConfiguration;
 import org.apache.asterix.runtime.writer.IExternalFileWriterFactory;
@@ -117,19 +122,20 @@
         Map<String, String> configuration = sink.getConfiguration();
         String format = configuration.get(ExternalDataConstants.KEY_FORMAT);
 
-        // Only JSON and parquet is supported for now
+        // Check for supported formats
         if (!ExternalDataConstants.FORMAT_JSON_LOWER_CASE.equalsIgnoreCase(format)
-                && !ExternalDataConstants.FORMAT_PARQUET.equalsIgnoreCase(format)) {
+                && !ExternalDataConstants.FORMAT_PARQUET.equalsIgnoreCase(format)
+                && !ExternalDataConstants.FORMAT_CSV_LOWER_CASE.equalsIgnoreCase(format)) {
             throw new UnsupportedOperationException("Unsupported format " + format);
         }
 
         String compression = getCompression(configuration);
-
+        IPrinterFactory printerFactory;
+        IExternalFileCompressStreamFactory compressStreamFactory;
         switch (format) {
             case ExternalDataConstants.FORMAT_JSON_LOWER_CASE:
-                IExternalFileCompressStreamFactory compressStreamFactory =
-                        createCompressionStreamFactory(appCtx, compression, configuration);
-                IPrinterFactory printerFactory = CleanJSONPrinterFactoryProvider.INSTANCE.getPrinterFactory(sourceType);
+                compressStreamFactory = createCompressionStreamFactory(appCtx, compression, configuration);
+                printerFactory = CleanJSONPrinterFactoryProvider.INSTANCE.getPrinterFactory(sourceType);
                 return new TextualExternalFilePrinterFactory(printerFactory, compressStreamFactory);
             case ExternalDataConstants.FORMAT_PARQUET:
                 String parquetSchemaString = configuration.get(ExternalDataConstants.PARQUET_SCHEMA_KEY);
@@ -151,6 +157,23 @@
 
                 return new ParquetExternalFilePrinterFactory(compressionCodecName, parquetSchemaString,
                         (IAType) sourceType, rowGroupSize, pageSize, writerVersion);
+            case ExternalDataConstants.FORMAT_CSV_LOWER_CASE:
+                compressStreamFactory = createCompressionStreamFactory(appCtx, compression, configuration);
+                if (sink instanceof IExternalWriteDataSink) {
+                    ARecordType itemType = ((IExternalWriteDataSink) sink).getItemType();
+                    if (itemType != null) {
+                        printerFactory =
+                                CSVPrinterFactoryProvider
+                                        .createInstance(itemType, sink.getConfiguration(),
+                                                ((IExternalWriteDataSink) sink).getSourceLoc())
+                                        .getPrinterFactory(sourceType);
+                        return new CsvExternalFilePrinterFactory(printerFactory, compressStreamFactory);
+                    } else {
+                        throw new CompilationException(ErrorCode.INVALID_CSV_SCHEMA);
+                    }
+                } else {
+                    throw new CompilationException(ErrorCode.INVALID_CSV_SCHEMA);
+                }
             default:
                 throw new UnsupportedOperationException("Unsupported format " + format);
         }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java
index c4da935..372b5fd 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java
@@ -18,6 +18,8 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.printers;
 
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.DEFAULT_QUOTE;
+
 import java.io.DataOutput;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -25,6 +27,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
@@ -280,29 +283,55 @@
         UPPER_CASE,
     }
 
-    public static void writeUTF8StringAsCSV(byte[] b, int s, int l, OutputStream os) throws IOException {
+    public static void writeUTF8StringAsCSV(byte[] b, int s, int l, PrintStream ps, char quote, boolean forceQuote,
+            char escape, char delimiter) throws IOException {
         int stringLength = UTF8StringUtil.getUTFLength(b, s);
         int position = s + UTF8StringUtil.getNumBytesToStoreLength(stringLength);
         int maxPosition = position + stringLength;
-        os.write('"');
+        char quoteChar = quote == CSVUtils.NULL_CHAR ? DEFAULT_QUOTE : quote;
+
+        boolean shouldQuote = forceQuote;
+        if (!shouldQuote) {
+            // Check if the string contains any special characters that require quoting
+            for (int i = position; i < maxPosition; i++) {
+                char c = UTF8StringUtil.charAt(b, i);
+                if (c == quote || c == '\r' || c == '\n' || c == escape || c == delimiter) {
+                    shouldQuote = true;
+                    break;
+                }
+            }
+        }
+
+        if (shouldQuote) {
+            ps.print(quoteChar);
+        }
+
         while (position < maxPosition) {
             char c = UTF8StringUtil.charAt(b, position);
             int sz = UTF8StringUtil.charSize(b, position);
-            if (c == '"') {
-                os.write('"');
+
+            // todo: Escape character handling -- as the data is strictly quoted in case of carriage return, should "\r" needs to get handled?
+            if (c == quote || c == '\r') {
+                ps.print(escape);
             }
+
+            // Handling surrogate pairs
             if (Character.isHighSurrogate(c)) {
-                position += writeSupplementaryChar(os, b, maxPosition, position, c, sz);
+                position += writeSupplementaryChar(ps, b, maxPosition, position, c, sz);
                 continue;
             }
+
+            // Write the character bytes
             while (sz > 0) {
-                os.write(b[position]);
+                ps.print(c);
                 ++position;
                 --sz;
             }
-            break;
         }
-        os.write('"');
+
+        if (shouldQuote) {
+            ps.print(quoteChar);
+        }
     }
 
     public static void writeUTF8StringRaw(byte[] b, int s, int l, DataOutput os) throws IOException {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ANullPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ANullPrinterFactory.java
index dcf98a0..3d3ca3e 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ANullPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ANullPrinterFactory.java
@@ -19,16 +19,34 @@
 package org.apache.asterix.dataflow.data.nontagged.printers.csv;
 
 import java.io.PrintStream;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.hyracks.algebricks.data.IPrinter;
 import org.apache.hyracks.algebricks.data.IPrinterFactory;
 
 public class ANullPrinterFactory implements IPrinterFactory {
-
     private static final long serialVersionUID = 1L;
-    public static final ANullPrinterFactory INSTANCE = new ANullPrinterFactory();
+    private static final String DEFAULT_NULL_STRING = "";
+    // Store the information about the instance based on the parameters
+    private static final ConcurrentHashMap<String, ANullPrinterFactory> instanceCache = new ConcurrentHashMap<>();
+    private String nullString;
 
-    public static final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> ps.print("null");
+    private ANullPrinterFactory(String nullString) {
+        this.nullString = nullString;
+    }
+
+    public static ANullPrinterFactory createInstance(String nullString) {
+        String key = CSVUtils.generateKey(nullString);
+        return instanceCache.computeIfAbsent(key, k -> new ANullPrinterFactory(nullString));
+    }
+
+    private final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
+        if (nullString != null) {
+            ps.print(nullString);
+        } else {
+            ps.print(DEFAULT_NULL_STRING);
+        }
+    };
 
     @Override
     public IPrinter createPrinter() {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AObjectPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AObjectPrinterFactory.java
index f1e2300..20b1ef0 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AObjectPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AObjectPrinterFactory.java
@@ -18,12 +18,22 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.printers.csv;
 
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_DELIMITER;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_EMPTY_FIELD_AS_NULL;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_ESCAPE;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_FORCE_QUOTE;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_NULL;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_QUOTE;
+
 import java.io.PrintStream;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.asterix.om.pointables.ARecordVisitablePointable;
 import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
 import org.apache.asterix.om.pointables.printer.IPrintVisitor;
 import org.apache.asterix.om.pointables.printer.csv.APrintVisitor;
+import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.EnumDeserializer;
 import org.apache.hyracks.algebricks.common.utils.Pair;
@@ -32,11 +42,26 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
 public class AObjectPrinterFactory implements IPrinterFactory {
-
     private static final long serialVersionUID = 1L;
-    public static final AObjectPrinterFactory INSTANCE = new AObjectPrinterFactory();
+    private static final ConcurrentHashMap<String, AObjectPrinterFactory> instanceCache = new ConcurrentHashMap<>();
+    private ARecordType itemType;
+    private Map<String, String> configuration;
+    private boolean emptyFieldAsNull;
 
-    public static boolean printFlatValue(ATypeTag typeTag, byte[] b, int s, int l, PrintStream ps)
+    private AObjectPrinterFactory(ARecordType itemType, Map<String, String> configuration) {
+        this.itemType = itemType;
+        this.configuration = configuration;
+        String emptyFieldAsNullStr = configuration.get(KEY_EMPTY_FIELD_AS_NULL);
+        this.emptyFieldAsNull = emptyFieldAsNullStr != null && Boolean.parseBoolean(emptyFieldAsNullStr);
+    }
+
+    public static AObjectPrinterFactory createInstance(ARecordType itemType, Map<String, String> configuration) {
+        // generate a unique identifier based on the parameters and hash the instance corresponding to it.
+        String key = CSVUtils.generateKey(itemType, configuration);
+        return instanceCache.computeIfAbsent(key, k -> new AObjectPrinterFactory(itemType, configuration));
+    }
+
+    public boolean printFlatValue(ATypeTag typeTag, byte[] b, int s, int l, PrintStream ps)
             throws HyracksDataException {
         switch (typeTag) {
             case TINYINT:
@@ -53,7 +78,7 @@
                 return true;
             case MISSING:
             case NULL:
-                ANullPrinterFactory.PRINTER.print(b, s, l, ps);
+                ANullPrinterFactory.createInstance(configuration.get(KEY_NULL)).createPrinter().print(b, s, l, ps);
                 return true;
             case BOOLEAN:
                 ABooleanPrinterFactory.PRINTER.print(b, s, l, ps);
@@ -104,7 +129,14 @@
                 ARectanglePrinterFactory.PRINTER.print(b, s, l, ps);
                 return true;
             case STRING:
-                AStringPrinterFactory.PRINTER.print(b, s, l, ps);
+                if (emptyFieldAsNull && CSVUtils.isEmptyString(b, s, l)) {
+                    ANullPrinterFactory.createInstance(configuration.get(KEY_NULL)).createPrinter().print(b, s, l, ps);
+                } else {
+                    AStringPrinterFactory
+                            .createInstance(configuration.get(KEY_QUOTE), configuration.get(KEY_FORCE_QUOTE),
+                                    configuration.get(KEY_ESCAPE), configuration.get(KEY_DELIMITER))
+                            .createPrinter().print(b, s, l, ps);
+                }
                 return true;
             case BINARY:
                 ABinaryHexPrinterFactory.PRINTER.print(b, s, l, ps);
@@ -119,11 +151,10 @@
 
     @Override
     public IPrinter createPrinter() {
-        final ARecordVisitablePointable rPointable =
+        final ARecordVisitablePointable recordVisitablePointable =
                 new ARecordVisitablePointable(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
         final Pair<PrintStream, ATypeTag> streamTag = new Pair<>(null, null);
-
-        final IPrintVisitor visitor = new APrintVisitor();
+        final IPrintVisitor visitor = new APrintVisitor(itemType, configuration);
 
         return (byte[] b, int s, int l, PrintStream ps) -> {
             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[s]);
@@ -132,8 +163,8 @@
                 streamTag.second = typeTag;
                 switch (typeTag) {
                     case OBJECT:
-                        rPointable.set(b, s, l);
-                        visitor.visit(rPointable, streamTag);
+                        recordVisitablePointable.set(b, s, l);
+                        visitor.visit(recordVisitablePointable, streamTag);
                         break;
                     default:
                         throw new HyracksDataException("No printer for type " + typeTag);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ARecordPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ARecordPrinterFactory.java
index 909fd60..2b31fb0 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ARecordPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/ARecordPrinterFactory.java
@@ -19,6 +19,7 @@
 package org.apache.asterix.dataflow.data.nontagged.printers.csv;
 
 import java.io.PrintStream;
+import java.util.Map;
 
 import org.apache.asterix.om.pointables.PointableAllocator;
 import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
@@ -36,9 +37,13 @@
 
     private static final long serialVersionUID = 1L;
     private final ARecordType recType;
+    private final ARecordType itemType;
+    private final Map<String, String> configuration;
 
-    public ARecordPrinterFactory(ARecordType recType) {
+    public ARecordPrinterFactory(ARecordType recType, ARecordType itemType, Map<String, String> configuration) {
         this.recType = recType;
+        this.itemType = itemType;
+        this.configuration = configuration;
     }
 
     @Override
@@ -47,7 +52,7 @@
         final IAType inputType =
                 recType == null ? DefaultOpenFieldType.getDefaultOpenFieldType(ATypeTag.OBJECT) : recType;
         final IVisitablePointable recAccessor = allocator.allocateRecordValue(inputType);
-        final APrintVisitor printVisitor = new APrintVisitor();
+        final APrintVisitor printVisitor = new APrintVisitor(itemType, configuration);
         final Pair<PrintStream, ATypeTag> arg = new Pair<>(null, null);
 
         return new IPrinter() {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AStringPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AStringPrinterFactory.java
index c217203..ae368bd 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AStringPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/AStringPrinterFactory.java
@@ -18,8 +18,14 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.printers.csv;
 
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.DEFAULT_VALUES;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_DELIMITER;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_ESCAPE;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_QUOTE;
+
 import java.io.IOException;
 import java.io.PrintStream;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.asterix.dataflow.data.nontagged.printers.PrintTools;
 import org.apache.hyracks.algebricks.data.IPrinter;
@@ -27,18 +33,52 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
 public class AStringPrinterFactory implements IPrinterFactory {
-
     private static final long serialVersionUID = 1L;
-    public static final AStringPrinterFactory INSTANCE = new AStringPrinterFactory();
+    private static final ConcurrentHashMap<String, AStringPrinterFactory> instanceCache = new ConcurrentHashMap<>();
+    private static final String NONE = "none";
+    private String quote;
+    private Boolean forceQuote;
+    private String escape;
+    private String delimiter;
 
-    public static final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
+    private AStringPrinterFactory(String quote, Boolean forceQuote, String escape, String delimiter) {
+        this.quote = quote;
+        this.forceQuote = forceQuote;
+        this.escape = escape;
+        this.delimiter = delimiter;
+    }
+
+    public static AStringPrinterFactory createInstance(String quote, String forceQuoteStr, String escape,
+            String delimiter) {
+        boolean forceQuote = forceQuoteStr == null || Boolean.parseBoolean(forceQuoteStr);
+        String key = CSVUtils.generateKey(quote, forceQuoteStr, escape, delimiter);
+        return instanceCache.computeIfAbsent(key, k -> new AStringPrinterFactory(quote, forceQuote, escape, delimiter));
+    }
+
+    private final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
         try {
-            PrintTools.writeUTF8StringAsCSV(b, s + 1, l - 1, ps);
+            char quoteChar =
+                    quote == null ? extractSingleChar(DEFAULT_VALUES.get(KEY_QUOTE)) : extractSingleChar(quote);
+            char escapeChar =
+                    escape == null ? extractSingleChar(DEFAULT_VALUES.get(KEY_ESCAPE)) : extractSingleChar(escape);
+            char delimiterChar = delimiter == null ? extractSingleChar(DEFAULT_VALUES.get(KEY_DELIMITER))
+                    : extractSingleChar(delimiter);
+            PrintTools.writeUTF8StringAsCSV(b, s + 1, l - 1, ps, quoteChar, forceQuote, escapeChar, delimiterChar);
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
     };
 
+    public char extractSingleChar(String input) throws IOException {
+        if (input != null && input.length() == 1) {
+            return input.charAt(0);
+        } else if (input.equalsIgnoreCase(NONE)) {
+            return CSVUtils.NULL_CHAR; // Replace 'none' with null character
+        } else {
+            throw new IOException("Input string must be a single character");
+        }
+    }
+
     @Override
     public IPrinter createPrinter() {
         return PRINTER;
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/CSVUtils.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/CSVUtils.java
new file mode 100644
index 0000000..b50816a
--- /dev/null
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/csv/CSVUtils.java
@@ -0,0 +1,102 @@
+/*
+ * 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.asterix.dataflow.data.nontagged.printers.csv;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.om.types.ARecordType;
+
+public class CSVUtils {
+
+    // Constants for the supported CSV parameters
+    public static final String KEY_NULL = "null";
+    public static final String KEY_ESCAPE = "escape";
+    public static final String KEY_HEADER = "header";
+    public static final String KEY_DELIMITER = "delimiter";
+    public static final String KEY_RECORD_DELIMITER = "recordDelimiter";
+    public static final String KEY_FORCE_QUOTE = "forceQuote";
+    public static final String KEY_QUOTE = "quote";
+    public static final String KEY_EMPTY_FIELD_AS_NULL = "empty_field_as_null";
+    public static final char DEFAULT_QUOTE = '"';
+    private static final String DEFAULT_DELIMITER_VALUE = ",";
+    private static final String DEFAULT_NULL_VALUE = "";
+    private static final String DOUBLE_QUOTES = "\"";
+    public static final char NULL_CHAR = '\0';
+    private static final String FALSE = "false";
+    private static final String DEFAULT_RECORD_DELIMITER = "\n";
+
+    // List of supported parameters
+    public static final List<String> CSV_PARAMETERS = Arrays.asList(KEY_NULL, KEY_ESCAPE, KEY_HEADER, KEY_DELIMITER,
+            KEY_RECORD_DELIMITER, KEY_FORCE_QUOTE, KEY_QUOTE, KEY_EMPTY_FIELD_AS_NULL);
+
+    // Default values for each parameter
+    public static final Map<String, String> DEFAULT_VALUES;
+
+    static {
+        DEFAULT_VALUES = new HashMap<>();
+        DEFAULT_VALUES.put(KEY_NULL, DEFAULT_NULL_VALUE);
+        DEFAULT_VALUES.put(KEY_ESCAPE, DOUBLE_QUOTES);
+        DEFAULT_VALUES.put(KEY_HEADER, FALSE);
+        DEFAULT_VALUES.put(KEY_DELIMITER, DEFAULT_DELIMITER_VALUE);
+        DEFAULT_VALUES.put(KEY_RECORD_DELIMITER, DEFAULT_RECORD_DELIMITER);
+        DEFAULT_VALUES.put(KEY_FORCE_QUOTE, FALSE);
+        DEFAULT_VALUES.put(KEY_QUOTE, DOUBLE_QUOTES);
+        DEFAULT_VALUES.put(KEY_EMPTY_FIELD_AS_NULL, FALSE);
+    }
+
+    // Generate a key based on configuration for ARecordType and parameters
+    public static String generateKey(ARecordType itemType, Map<String, String> configuration) {
+        StringBuilder keyBuilder = new StringBuilder();
+        keyBuilder.append(itemType == null ? KEY_NULL : itemType.toString()).append(" | ");
+        // Iterate through supported CSV parameters and append their values from configuration
+        for (String param : CSV_PARAMETERS) {
+            String value = configuration.getOrDefault(param, DEFAULT_VALUES.get(param));
+            keyBuilder.append(param).append(" : ").append(value).append(" | ");
+        }
+        // Remove the trailing " | "
+        if (keyBuilder.length() > 0 && keyBuilder.charAt(keyBuilder.length() - 2) == '|') {
+            keyBuilder.setLength(keyBuilder.length() - 3);
+        }
+        return keyBuilder.toString();
+    }
+
+    public static String generateKey(String quote, String forceQuoteStr, String escape, String delimiter) {
+        // Use default values when no values are specified (null)
+        return KEY_QUOTE + " : " + (quote != null ? quote : DEFAULT_VALUES.get(KEY_QUOTE)) + " | " + KEY_FORCE_QUOTE
+                + " : " + (forceQuoteStr != null ? forceQuoteStr : DEFAULT_VALUES.get(KEY_FORCE_QUOTE)) + " | "
+                + KEY_ESCAPE + " : " + (escape != null ? escape : DEFAULT_VALUES.get(KEY_ESCAPE)) + " | "
+                + KEY_DELIMITER + " : " + (delimiter != null ? delimiter : DEFAULT_VALUES.get(KEY_DELIMITER));
+    }
+
+    public static String generateKey(String nullString) {
+        // Use the default value when nullString is not specified (null)
+        return KEY_NULL + " : " + (nullString != null ? nullString : DEFAULT_VALUES.get(KEY_NULL));
+    }
+
+    public static boolean isEmptyString(byte[] b, int s, int l) {
+        return b == null || l <= 2 || s < 0 || s + l > b.length;
+    }
+
+    public static String getDelimiter(Map<String, String> configuration) {
+        return configuration.get(KEY_DELIMITER) == null ? DEFAULT_DELIMITER_VALUE : configuration.get(KEY_DELIMITER);
+    }
+}
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/formats/nontagged/CSVPrinterFactoryProvider.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/formats/nontagged/CSVPrinterFactoryProvider.java
index b8201ae..322b3e6 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/formats/nontagged/CSVPrinterFactoryProvider.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/formats/nontagged/CSVPrinterFactoryProvider.java
@@ -18,6 +18,15 @@
  */
 package org.apache.asterix.formats.nontagged;
 
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_DELIMITER;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_ESCAPE;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_FORCE_QUOTE;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_NULL;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_QUOTE;
+
+import java.util.Collections;
+import java.util.Map;
+
 import org.apache.asterix.dataflow.data.nontagged.printers.adm.ShortWithoutTypeInfoPrinterFactory;
 import org.apache.asterix.dataflow.data.nontagged.printers.csv.ABooleanPrinterFactory;
 import org.apache.asterix.dataflow.data.nontagged.printers.csv.ACirclePrinterFactory;
@@ -52,12 +61,26 @@
 import org.apache.hyracks.algebricks.common.exceptions.NotImplementedException;
 import org.apache.hyracks.algebricks.data.IPrinterFactory;
 import org.apache.hyracks.algebricks.data.IPrinterFactoryProvider;
+import org.apache.hyracks.api.exceptions.SourceLocation;
 
 public class CSVPrinterFactoryProvider implements IPrinterFactoryProvider {
+    private ARecordType itemType;
+    private Map<String, String> configuration;
+    private SourceLocation sourceLocation;
 
-    public static final CSVPrinterFactoryProvider INSTANCE = new CSVPrinterFactoryProvider();
+    public static final CSVPrinterFactoryProvider INSTANCE =
+            new CSVPrinterFactoryProvider(null, Collections.emptyMap(), null);
 
-    private CSVPrinterFactoryProvider() {
+    public static final CSVPrinterFactoryProvider createInstance(ARecordType itemType,
+            Map<String, String> configuration, SourceLocation sourceLocation) {
+        return new CSVPrinterFactoryProvider(itemType, configuration, sourceLocation);
+    }
+
+    private CSVPrinterFactoryProvider(ARecordType itemType, Map<String, String> configuration,
+            SourceLocation sourceLocation) {
+        this.itemType = itemType;
+        this.configuration = configuration;
+        this.sourceLocation = sourceLocation;
     }
 
     @Override
@@ -76,7 +99,7 @@
                     return AInt64PrinterFactory.INSTANCE;
                 case MISSING:
                 case NULL:
-                    return ANullPrinterFactory.INSTANCE;
+                    ANullPrinterFactory.createInstance(configuration.get(KEY_NULL));
                 case BOOLEAN:
                     return ABooleanPrinterFactory.INSTANCE;
                 case FLOAT:
@@ -110,13 +133,15 @@
                 case RECTANGLE:
                     return ARectanglePrinterFactory.INSTANCE;
                 case STRING:
-                    return AStringPrinterFactory.INSTANCE;
+                    return AStringPrinterFactory.createInstance(configuration.get(KEY_QUOTE),
+                            configuration.get(KEY_FORCE_QUOTE), configuration.get(KEY_ESCAPE),
+                            configuration.get(KEY_DELIMITER));
                 case OBJECT:
-                    return new ARecordPrinterFactory((ARecordType) type);
+                    return new ARecordPrinterFactory((ARecordType) type, itemType, configuration);
                 case ARRAY:
-                    throw new NotImplementedException("'Orderedlist' type unsupported for CSV output");
+                    throw new NotImplementedException("'OrderedList' type unsupported for CSV output");
                 case MULTISET:
-                    throw new NotImplementedException("'Unorderedlist' type unsupported for CSV output");
+                    throw new NotImplementedException("'UnorderedList' type unsupported for CSV output");
                 case UNION:
                     if (((AUnionType) type).isUnknownableType()) {
                         return new AOptionalFieldPrinterFactory((AUnionType) type);
@@ -142,7 +167,7 @@
                     break;
             }
         }
-        return AObjectPrinterFactory.INSTANCE;
+        return AObjectPrinterFactory.createInstance(itemType, configuration);
 
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/ARecordPrinter.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/ARecordPrinter.java
index 34d618d..9aaf889 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/ARecordPrinter.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/ARecordPrinter.java
@@ -33,13 +33,13 @@
  * This class is to print the content of a record.
  */
 public class ARecordPrinter {
-    private final String startRecord;
-    private final String endRecord;
-    private final String fieldSeparator;
-    private final String fieldNameSeparator;
+    protected final String startRecord;
+    protected final String endRecord;
+    protected final String fieldSeparator;
+    protected final String fieldNameSeparator;
 
-    private final Pair<PrintStream, ATypeTag> nameVisitorArg = new Pair<>(null, ATypeTag.STRING);
-    private final Pair<PrintStream, ATypeTag> itemVisitorArg = new Pair<>(null, null);
+    protected final Pair<PrintStream, ATypeTag> nameVisitorArg = new Pair<>(null, ATypeTag.STRING);
+    protected final Pair<PrintStream, ATypeTag> itemVisitorArg = new Pair<>(null, null);
 
     public ARecordPrinter(final String startRecord, final String endRecord, final String fieldSeparator,
             final String fieldNameSeparator) {
@@ -82,7 +82,7 @@
         ps.print(endRecord);
     }
 
-    private void printField(PrintStream ps, IPrintVisitor visitor, IVisitablePointable fieldName,
+    protected void printField(PrintStream ps, IPrintVisitor visitor, IVisitablePointable fieldName,
             IVisitablePointable fieldValue, ATypeTag fieldTypeTag) throws HyracksDataException {
         itemVisitorArg.second = fieldTypeTag;
         if (fieldNameSeparator != null) {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/ACSVRecordPrinter.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/ACSVRecordPrinter.java
new file mode 100644
index 0000000..ce9e3a3
--- /dev/null
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/ACSVRecordPrinter.java
@@ -0,0 +1,182 @@
+/*
+ * 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.asterix.om.pointables.printer.csv;
+
+import static org.apache.asterix.om.types.hierachy.ATypeHierarchy.isCompatible;
+
+import java.io.PrintStream;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.om.pointables.ARecordVisitablePointable;
+import org.apache.asterix.om.pointables.base.IVisitablePointable;
+import org.apache.asterix.om.pointables.printer.ARecordPrinter;
+import org.apache.asterix.om.pointables.printer.IPrintVisitor;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.AUnionType;
+import org.apache.asterix.om.types.EnumDeserializer;
+import org.apache.asterix.om.types.IAType;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.util.string.UTF8StringUtil;
+
+public class ACSVRecordPrinter extends ARecordPrinter {
+    private ARecordType schema;
+    private boolean firstRecord;
+    private boolean header;
+    private final String recordDelimiter;
+    private static final List<ATypeTag> supportedTypes = List.of(ATypeTag.TINYINT, ATypeTag.SMALLINT, ATypeTag.INTEGER,
+            ATypeTag.BIGINT, ATypeTag.UINT8, ATypeTag.UINT16, ATypeTag.UINT64, ATypeTag.FLOAT, ATypeTag.DOUBLE,
+            ATypeTag.STRING, ATypeTag.BOOLEAN, ATypeTag.DATETIME, ATypeTag.UINT32, ATypeTag.DATE, ATypeTag.TIME);
+
+    public ACSVRecordPrinter(final String startRecord, final String endRecord, final String fieldSeparator,
+            final String fieldNameSeparator, String recordDelimiter, ARecordType schema, String headerStr) {
+        super(startRecord, endRecord, fieldSeparator, fieldNameSeparator);
+        this.schema = schema;
+        this.header = headerStr != null && Boolean.parseBoolean(headerStr);
+        this.firstRecord = true;
+        this.recordDelimiter = recordDelimiter;
+    }
+
+    @Override
+    public void printRecord(ARecordVisitablePointable recordAccessor, PrintStream ps, IPrintVisitor visitor)
+            throws HyracksDataException {
+        // backward compatibility -- No Schema print it as it is from recordAccessor
+        if (schema == null) {
+            super.printRecord(recordAccessor, ps, visitor);
+        } else {
+            printSchemaFullRecord(recordAccessor, ps, visitor);
+        }
+    }
+
+    private void printSchemaFullRecord(ARecordVisitablePointable recordAccessor, PrintStream ps, IPrintVisitor visitor)
+            throws HyracksDataException {
+        // check the schema for the record
+        // try producing the record into the record of expected schema
+        Map<String, ATypeTag> schemaDetails = new HashMap<>();
+        if (checkCSVSchema(recordAccessor, schemaDetails)) {
+            nameVisitorArg.first = ps;
+            itemVisitorArg.first = ps;
+            if (header) {
+                addHeader(recordAccessor, ps, visitor);
+            }
+            // add record delimiter
+            // by default the separator between the header and the records is "\n"
+            if (firstRecord) {
+                firstRecord = false;
+            } else {
+                ps.print(recordDelimiter);
+            }
+            final List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+            final List<IVisitablePointable> fieldValues = recordAccessor.getFieldValues();
+
+            boolean first = true;
+            for (int i = 0; i < fieldNames.size(); ++i) {
+                final IVisitablePointable fieldNamePointable = fieldNames.get(i);
+                String fieldName = UTF8StringUtil.toString(fieldNamePointable.getByteArray(),
+                        fieldNamePointable.getStartOffset() + 1);
+                final IVisitablePointable fieldValue = fieldValues.get(i);
+                final ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
+                        .deserialize(fieldValue.getByteArray()[fieldValue.getStartOffset()]);
+                ATypeTag expectedTypeTag = schemaDetails.get(fieldName);
+                if (!isCompatible(typeTag, expectedTypeTag)) {
+                    expectedTypeTag = ATypeTag.NULL;
+                }
+                if (first) {
+                    first = false;
+                } else {
+                    ps.print(fieldSeparator);
+                }
+                printField(ps, visitor, fieldNamePointable, fieldValue, expectedTypeTag);
+            }
+        }
+    }
+
+    private boolean checkCSVSchema(ARecordVisitablePointable recordAccessor, Map<String, ATypeTag> schemaDetails) {
+        final List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+        final List<IVisitablePointable> fieldValues = recordAccessor.getFieldValues();
+        final List<String> expectedFieldNames = Arrays.asList(schema.getFieldNames());
+        final List<IAType> expectedFieldTypes = Arrays.asList(schema.getFieldTypes());
+        if (fieldNames.size() != expectedFieldNames.size()) {
+            // todo: raise warning about schema mismatch
+            return false;
+        }
+        for (int i = 0; i < fieldNames.size(); ++i) {
+            final IVisitablePointable fieldName = fieldNames.get(i);
+            String fieldColumnName = UTF8StringUtil.toString(fieldName.getByteArray(), fieldName.getStartOffset() + 1);
+            final IVisitablePointable fieldValue = fieldValues.get(i);
+            final ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
+                    .deserialize(fieldValue.getByteArray()[fieldValue.getStartOffset()]);
+            ATypeTag expectedType;
+            boolean canNull = false;
+            if (expectedFieldNames.contains(fieldColumnName)) {
+                IAType expectedIAType = expectedFieldTypes.get(expectedFieldNames.indexOf(fieldColumnName));
+                if (!supportedTypes.contains(expectedIAType.getTypeTag())) {
+                    if (expectedIAType.getTypeTag().equals(ATypeTag.UNION)) {
+                        AUnionType unionType = (AUnionType) expectedIAType;
+                        expectedType = unionType.getActualType().getTypeTag();
+                        canNull = unionType.isNullableType();
+                        if (!supportedTypes.contains(expectedType)) {
+                            // unsupported DataType
+                            return false;
+                        }
+                    } else {
+                        // todo: unexpected type
+                        return false;
+                    }
+                } else {
+                    expectedType = expectedIAType.getTypeTag();
+                }
+                schemaDetails.put(fieldColumnName, expectedType);
+            } else {
+                // todo: raise warning about schema mismatch
+                return false;
+            }
+            if (typeTag.equals(ATypeTag.MISSING) || (typeTag.equals(ATypeTag.NULL) && !canNull)) {
+                // todo: raise warning about schema mismatch
+                return false;
+            }
+            if (!isCompatible(typeTag, expectedType) && !canNull) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private void addHeader(ARecordVisitablePointable recordAccessor, PrintStream ps, IPrintVisitor visitor)
+            throws HyracksDataException {
+        //check if it is a first record
+        if (firstRecord) {
+            final List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+            boolean first = true;
+            for (int i = 0; i < fieldNames.size(); ++i) {
+                if (first) {
+                    first = false;
+                } else {
+                    ps.print(fieldSeparator);
+                }
+                printFieldName(ps, visitor, fieldNames.get(i));
+            }
+            firstRecord = false;
+        }
+    }
+}
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/APrintVisitor.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/APrintVisitor.java
index 3f22374..22c502b 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/APrintVisitor.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/printer/csv/APrintVisitor.java
@@ -19,14 +19,20 @@
 
 package org.apache.asterix.om.pointables.printer.csv;
 
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_HEADER;
+import static org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils.KEY_RECORD_DELIMITER;
+
 import java.io.PrintStream;
+import java.util.Map;
 
 import org.apache.asterix.dataflow.data.nontagged.printers.csv.AObjectPrinterFactory;
+import org.apache.asterix.dataflow.data.nontagged.printers.csv.CSVUtils;
 import org.apache.asterix.om.pointables.AListVisitablePointable;
 import org.apache.asterix.om.pointables.ARecordVisitablePointable;
 import org.apache.asterix.om.pointables.printer.AListPrinter;
 import org.apache.asterix.om.pointables.printer.ARecordPrinter;
 import org.apache.asterix.om.pointables.printer.AbstractPrintVisitor;
+import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
@@ -36,6 +42,15 @@
  * PrintStream in CSV format.
  */
 public class APrintVisitor extends AbstractPrintVisitor {
+    private final ARecordType itemType;
+    private final Map<String, String> configuration;
+
+    public APrintVisitor(ARecordType itemType, Map<String, String> configuration) {
+        super();
+        this.itemType = itemType;
+        this.configuration = configuration;
+    }
+
     @Override
     protected AListPrinter createListPrinter(AListVisitablePointable accessor) throws HyracksDataException {
         throw new HyracksDataException("'List' type unsupported for CSV output");
@@ -43,12 +58,15 @@
 
     @Override
     protected ARecordPrinter createRecordPrinter(ARecordVisitablePointable accessor) {
-        return new ARecordPrinter("", "", ",", null);
+        String delimiter = CSVUtils.getDelimiter(configuration);
+        String recordDelimiter = configuration.get(KEY_RECORD_DELIMITER) == null ? (itemType == null ? "" : "\n")
+                : configuration.get(KEY_RECORD_DELIMITER);
+        return new ACSVRecordPrinter("", "", delimiter, null, recordDelimiter, itemType, configuration.get(KEY_HEADER));
     }
 
     @Override
     protected boolean printFlatValue(ATypeTag typeTag, byte[] b, int s, int l, PrintStream ps)
             throws HyracksDataException {
-        return AObjectPrinterFactory.printFlatValue(typeTag, b, s, l, ps);
+        return AObjectPrinterFactory.createInstance(itemType, configuration).printFlatValue(typeTag, b, s, l, ps);
     }
 }