diff --git a/asterix-app/pom.xml b/asterix-app/pom.xml
index cad13ce..25f1ad9 100644
--- a/asterix-app/pom.xml
+++ b/asterix-app/pom.xml
@@ -219,11 +219,13 @@
             <groupId>org.mockito</groupId>
             <artifactId>mockito-all</artifactId>
             <version>1.10.19</version>
+            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>com.e-movimento.tinytools</groupId>
             <artifactId>privilegedaccessor</artifactId>
             <version>1.2.2</version>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServlet.java b/asterix-app/src/main/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServlet.java
index 7559326..267aac5 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServlet.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServlet.java
@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.util.List;
 import java.util.Map;
 
 import javax.servlet.ServletContext;
@@ -33,6 +34,8 @@
 import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
 import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
 import edu.uci.ics.asterix.metadata.entities.Dataset;
+import edu.uci.ics.asterix.metadata.utils.DatasetUtils;
+import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.hyracks.api.client.IHyracksClientConnection;
 import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
 import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
@@ -90,9 +93,18 @@
             boolean temp = dataset.getDatasetDetails().isTemp();
             FileSplit[] fileSplits = metadataProvider.splitsForDataset(mdTxnCtx, dataverseName, datasetName,
                     datasetName, temp);
+            ARecordType recordType = (ARecordType) metadataProvider.findType(dataverseName, dataset.getItemTypeName());
+            List<List<String>> primaryKeys = DatasetUtils.getPartitioningKeys(dataset);
+            StringBuilder pkStrBuf = new StringBuilder();
+            for (List<String> keys : primaryKeys) {
+                for (String key : keys) {
+                    pkStrBuf.append(key).append(",");
+                }
+            }
+            pkStrBuf.delete(pkStrBuf.length() - 1, pkStrBuf.length());
 
             // Constructs the returned json object.
-            formResponseObject(jsonResponse, fileSplits, hcc.getNodeControllerInfos());
+            formResponseObject(jsonResponse, fileSplits, recordType, pkStrBuf.toString(), hcc.getNodeControllerInfos());
             // Metadata transaction commits.
             MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
             // Writes file splits.
@@ -106,9 +118,13 @@
         }
     }
 
-    private void formResponseObject(JSONObject jsonResponse, FileSplit[] fileSplits,
-            Map<String, NodeControllerInfo> nodeMap) throws Exception {
+    private void formResponseObject(JSONObject jsonResponse, FileSplit[] fileSplits, ARecordType recordType,
+            String primaryKeys, Map<String, NodeControllerInfo> nodeMap) throws Exception {
         JSONArray partititons = new JSONArray();
+        // Adds a primary key.
+        jsonResponse.put("keys", primaryKeys);
+        // Adds record type.
+        jsonResponse.put("type", recordType.toJSON());
         // Generates file partitions.
         for (FileSplit split : fileSplits) {
             String ipAddress = nodeMap.get(split.getNodeName()).getNetworkAddress().getAddress().toString();
diff --git a/asterix-app/src/test/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServletTest.java b/asterix-app/src/test/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServletTest.java
index bf651ad..7f5f480 100644
--- a/asterix-app/src/test/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServletTest.java
+++ b/asterix-app/src/test/java/edu/uci/ics/asterix/api/http/servlet/ConnectorAPIServletTest.java
@@ -36,17 +36,26 @@
 import junit.framework.Assert;
 
 import org.json.JSONArray;
-import org.json.JSONException;
 import org.json.JSONObject;
 import org.json.JSONTokener;
 import org.junit.Test;
 
+import edu.uci.ics.asterix.feeds.CentralFeedManager;
+import edu.uci.ics.asterix.metadata.MetadataManager;
+import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
+import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
+import edu.uci.ics.asterix.metadata.entities.Dataset;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.JSONDeserializerForTypes;
 import edu.uci.ics.asterix.test.runtime.ExecutionTest;
 import edu.uci.ics.hyracks.api.client.IHyracksClientConnection;
 import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
 import edu.uci.ics.hyracks.api.comm.NetworkAddress;
 import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
 
+@SuppressWarnings("deprecation")
 public class ConnectorAPIServletTest {
 
     @Test
@@ -90,6 +99,13 @@
                 new ByteArrayInputStream(outputStream.toByteArray())));
         JSONObject actualResponse = new JSONObject(tokener);
 
+        // Checks the data type of the dataset.
+        String primaryKey = actualResponse.getString("keys");
+        Assert.assertEquals("DataverseName,DatasetName", primaryKey);
+        ARecordType recordType = (ARecordType) JSONDeserializerForTypes.convertFromJSON((JSONObject) actualResponse
+                .get("type"));
+        Assert.assertEquals(getMetadataRecordType("Metadata", "Dataset"), recordType);
+
         // Checks the correctness of results.
         JSONArray splits = actualResponse.getJSONArray("splits");
         String path = ((JSONObject) splits.get(0)).getString("path");
@@ -100,7 +116,7 @@
     }
 
     @Test
-    public void testFormResponseObject() throws JSONException {
+    public void testFormResponseObject() throws Exception {
         ConnectorAPIServlet servlet = new ConnectorAPIServlet();
         JSONObject actualResponse = new JSONObject();
         FileSplit[] splits = new FileSplit[2];
@@ -114,15 +130,23 @@
         when(mockInfo1.getNetworkAddress()).thenReturn(new NetworkAddress("127.0.0.1", 3099));
         when(mockInfo2.getNetworkAddress()).thenReturn(new NetworkAddress("127.0.0.2", 3099));
 
+        String[] fieldNames = new String[] { "a1", "a2" };
+        IAType[] fieldTypes = new IAType[] { BuiltinType.ABOOLEAN, BuiltinType.ADAYTIMEDURATION };
+        ARecordType recordType = new ARecordType("record", fieldNames, fieldTypes, true);
+        String primaryKey = "a1";
+
         // Calls ConnectorAPIServlet.formResponseObject.
         nodeMap.put("nc1", mockInfo1);
         nodeMap.put("nc2", mockInfo2);
         PA.invokeMethod(servlet,
                 "formResponseObject(org.json.JSONObject, edu.uci.ics.hyracks.dataflow.std.file.FileSplit[], "
-                        + "java.util.Map)", actualResponse, splits, nodeMap);
+                        + "edu.uci.ics.asterix.om.types.ARecordType, java.lang.String, java.util.Map)", actualResponse,
+                splits, recordType, primaryKey, nodeMap);
 
         // Constructs expected response.
         JSONObject expectedResponse = new JSONObject();
+        expectedResponse.put("keys", primaryKey);
+        expectedResponse.put("type", recordType.toJSON());
         JSONArray splitsArray = new JSONArray();
         JSONObject element1 = new JSONObject();
         element1.put("ip", "127.0.0.1");
@@ -137,4 +161,16 @@
         // Checks results.
         Assert.assertEquals(actualResponse.toString(), expectedResponse.toString());
     }
+
+    private ARecordType getMetadataRecordType(String dataverseName, String datasetName) throws Exception {
+        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+        // Retrieves file splits of the dataset.
+        AqlMetadataProvider metadataProvider = new AqlMetadataProvider(null, CentralFeedManager.getInstance());
+        metadataProvider.setMetadataTxnContext(mdTxnCtx);
+        Dataset dataset = metadataProvider.findDataset(dataverseName, datasetName);
+        ARecordType recordType = (ARecordType) metadataProvider.findType(dataverseName, dataset.getItemTypeName());
+        // Metadata transaction commits.
+        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+        return recordType;
+    }
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AOrderedListType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AOrderedListType.java
index c1412da..147c0f3 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AOrderedListType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AOrderedListType.java
@@ -50,7 +50,7 @@
     public boolean equals(Object obj) {
         if (obj instanceof AOrderedListType) {
             AOrderedListType type = (AOrderedListType) obj;
-            return this.itemType == type.itemType;
+            return this.itemType.equals(type.itemType);
         }
         return false;
     }
@@ -73,7 +73,8 @@
     @Override
     public JSONObject toJSON() throws JSONException {
         JSONObject type = new JSONObject();
-        type.put("type", itemType);
+        type.put("type", AOrderedListType.class.getName());
+        type.put("item-type", itemType.toJSON());
         return type;
     }
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
index 826182f..f7e0a19 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
@@ -130,7 +130,7 @@
 
     /**
      * Returns the position of the field in the closed schema or -1 if the field does not exist.
-     * 
+     *
      * @param bytes
      *            the serialized bytes of the field name
      * @param start
@@ -209,7 +209,7 @@
 
     /**
      * Returns the position of the field in the closed schema or -1 if the field does not exist.
-     * 
+     *
      * @param fieldName
      *            the name of the field whose position is sought
      * @return the position of the field in the closed schema or -1 if the field does not exist.
@@ -267,7 +267,7 @@
 
     /**
      * Returns the field type of the field name if it exists, otherwise null.
-     * 
+     *
      * @param fieldName
      *            the fieldName whose type is sought
      * @return the field type of the field name if it exists, otherwise null
@@ -284,7 +284,7 @@
 
     /**
      * Returns true or false indicating whether or not a field is closed.
-     * 
+     *
      * @param fieldName
      *            the name of the field to check
      * @return true if fieldName is a closed field, otherwise false
@@ -296,7 +296,7 @@
 
     /**
      * Validates the partitioning expression that will be used to partition a dataset and returns expression type.
-     * 
+     *
      * @param partitioningExprs
      *            a list of partitioning expressions that will be validated
      * @return a list of partitioning expressions types
@@ -368,7 +368,7 @@
 
     /**
      * Validates the key fields that will be used as keys of an index.
-     * 
+     *
      * @param keyFieldNames
      *            a map of key fields that will be validated
      * @param keyFieldTypes
@@ -488,7 +488,7 @@
 
     /**
      * Validates the field that will be used as filter for the components of an LSM index.
-     * 
+     *
      * @param keyFieldNames
      *            a list of key fields that will be validated
      * @param indexType
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnionType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnionType.java
index 6399748..21e24fe 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnionType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnionType.java
@@ -122,10 +122,9 @@
     @Override
     public JSONObject toJSON() throws JSONException {
         JSONObject type = new JSONObject();
-        type.put("type", "UNION");
+        type.put("type", AUnionType.class.getName());
 
         JSONArray fields = new JSONArray();
-
         Iterator<IAType> iter = unionList.iterator();
         if (iter.hasNext()) {
             IAType t0 = iter.next();
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnorderedListType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnorderedListType.java
index 65244ae..93909bc 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnorderedListType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/AUnorderedListType.java
@@ -50,7 +50,7 @@
     public boolean equals(Object obj) {
         if (obj instanceof AUnorderedListType) {
             AUnorderedListType type = (AUnorderedListType) obj;
-            return this.itemType == type.itemType;
+            return this.itemType.equals(type.itemType);
         }
         return false;
     }
@@ -73,7 +73,8 @@
     @Override
     public JSONObject toJSON() throws JSONException {
         JSONObject type = new JSONObject();
-        type.put("type", itemType);
+        type.put("type", AUnorderedListType.class.getName());
+        type.put("item-type", itemType.toJSON());
         return type;
     }
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/BuiltinType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/BuiltinType.java
index da63b06..f0728db 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/BuiltinType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/BuiltinType.java
@@ -51,7 +51,7 @@
 
         @Override
         public String getTypeName() {
-            return "atype";
+            return "ASTERIX_TYPE";
         }
 
         @Override
@@ -62,7 +62,7 @@
         @Override
         public JSONObject toJSON() throws JSONException {
             JSONObject type = new JSONObject();
-            type.put("type", "AsterixType");
+            type.put("type", "ASTERIX_TYPE");
             return type;
         }
     };
@@ -305,7 +305,7 @@
         @Override
         public JSONObject toJSON() throws JSONException {
             JSONObject type = new JSONObject();
-            type.put("type", "Null");
+            type.put("type", "ANULL");
             return type;
         }
     };
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypes.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypes.java
new file mode 100644
index 0000000..4139fb1
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypes.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed 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 from
+ *
+ *     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 edu.uci.ics.asterix.om.util;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+
+public class JSONDeserializerForTypes {
+
+    /**
+     * Deserialize an arbitrary JSON representation of a type.
+     *
+     * @param typeInJSON
+     *            the JSON representation of the type.
+     * @return an valid AsterixDB type.
+     * @throws Exception
+     */
+    public static IAType convertFromJSON(JSONObject typeInJSON) throws Exception {
+        boolean typeNameExists = typeInJSON.has("type");
+        String typeName = typeNameExists ? typeInJSON.getString("type") : null;
+
+        // Deals with ordered list.
+        if (typeNameExists && typeName.equals(AOrderedListType.class.getName())) {
+            IAType itemType = convertFromJSON((JSONObject) typeInJSON.get("item-type"));
+            return new AOrderedListType(itemType, "ordered-list");
+        }
+
+        // Deals with unordered list.
+        if (typeNameExists && typeName.equals(AUnorderedListType.class.getName())) {
+            IAType itemType = convertFromJSON((JSONObject) typeInJSON.get("item-type"));
+            return new AUnorderedListType(itemType, "unordered-list");
+        }
+
+        // Deals with Union Type.
+        if (typeNameExists && typeName.equals(AUnionType.class.getName())) {
+            List<IAType> unionTypes = new ArrayList<IAType>();
+            JSONArray fields = (JSONArray) typeInJSON.get("fields");
+            for (int i = 0; i < fields.length(); i++) {
+                JSONObject fieldType = (JSONObject) fields.get(i);
+                unionTypes.add(convertFromJSON(fieldType));
+            }
+            return new AUnionType(unionTypes, "union");
+        }
+
+        // Deals with primitive types.
+        if (typeNameExists) {
+            Class<?> cl = BuiltinType.class;
+            Field typeField = cl.getDeclaredField(typeName.toUpperCase());
+            return (IAType) typeField.get(null);
+        }
+
+        // Deals with record types.
+        boolean openType = typeInJSON.getBoolean("open");
+        JSONArray fields = typeInJSON.getJSONArray("fields");
+        String[] fieldNames = new String[fields.length()];
+        IAType[] fieldTypes = new IAType[fields.length()];
+        for (int i = 0; i < fields.length(); ++i) {
+            JSONObject field = (JSONObject) fields.get(i);
+            JSONArray names = field.names();
+            String fieldName = names.getString(0);
+            fieldNames[i] = fieldName;
+            fieldTypes[i] = convertFromJSON((JSONObject) field.get(fieldName));
+        }
+        return new ARecordType("record", fieldNames, fieldTypes, openType);
+    }
+
+}
diff --git a/asterix-om/src/test/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypesTest.java b/asterix-om/src/test/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypesTest.java
new file mode 100644
index 0000000..53c81cc
--- /dev/null
+++ b/asterix-om/src/test/java/edu/uci/ics/asterix/om/util/JSONDeserializerForTypesTest.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ *  Licensed 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 from
+ *
+ *     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 edu.uci.ics.asterix.om.util;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.junit.Test;
+
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.JSONDeserializerForTypes;
+
+@SuppressWarnings("deprecation")
+public class JSONDeserializerForTypesTest {
+
+    @Test
+    public void test() throws Exception {
+        // Tests a record type with primitive types.
+        String[] fieldNames = { "a1", "a2", "a3" };
+        IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.AINT16, BuiltinType.ABITARRAY };
+        ARecordType recordType = new ARecordType("ARecord", fieldNames, fieldTypes, true);
+        Assert.assertEquals(recordType, JSONDeserializerForTypes.convertFromJSON(recordType.toJSON()));
+
+        // Tests a record type with a nested record type.
+        String[] fieldNames2 = { "a1", "a2" };
+        IAType[] fieldTypes2 = { BuiltinType.ABOOLEAN, recordType };
+        ARecordType recordType2 = new ARecordType("ARecord2", fieldNames2, fieldTypes2, true);
+        Assert.assertEquals(recordType2, JSONDeserializerForTypes.convertFromJSON(recordType2.toJSON()));
+
+        // Tests a record type with a union type.
+        String[] fieldNames3 = { "a1", "a2" };
+        List<IAType> unionTypes = new ArrayList<IAType>();
+        unionTypes.add(BuiltinType.ADURATION);
+        unionTypes.add(recordType2);
+        AUnionType unionType = new AUnionType(unionTypes, "union");
+        IAType[] fieldTypes3 = { BuiltinType.ABOOLEAN, unionType };
+        ARecordType recordType3 = new ARecordType("ARecord3", fieldNames3, fieldTypes3, true);
+        Assert.assertEquals(recordType3, JSONDeserializerForTypes.convertFromJSON(recordType3.toJSON()));
+
+        // Tests a record type with an ordered list.
+        String[] fieldNames4 = { "a1", "a2" };
+        IAType[] fieldTypes4 = { BuiltinType.ABOOLEAN, new AOrderedListType(BuiltinType.ADATETIME, "list") };
+        ARecordType recordType4 = new ARecordType("ARecord4", fieldNames4, fieldTypes4, false);
+        Assert.assertEquals(recordType4, JSONDeserializerForTypes.convertFromJSON(recordType4.toJSON()));
+
+        // Tests a record type with an unordered list.
+        String[] fieldNames5 = { "a1", "a2" };
+        IAType[] fieldTypes5 = { BuiltinType.ABOOLEAN, new AUnorderedListType(recordType2, "list") };
+        ARecordType recordType5 = new ARecordType("ARecord5", fieldNames5, fieldTypes5, false);
+        Assert.assertEquals(recordType5, JSONDeserializerForTypes.convertFromJSON(recordType5.toJSON()));
+    }
+}