diff --git a/server/src/main/java/io/druid/metadata/SQLMetadataConnector.java b/server/src/main/java/io/druid/metadata/SQLMetadataConnector.java
index 07b123003fa..73976e2ac48 100644
--- a/server/src/main/java/io/druid/metadata/SQLMetadataConnector.java
+++ b/server/src/main/java/io/druid/metadata/SQLMetadataConnector.java
@@ -373,13 +373,13 @@ public abstract class SQLMetadataConnector implements MetadataStorageConnector
         ImmutableList.of(
             String.format(
                 "CREATE TABLE %1$s (\n"
-                + "  id VARCHAR(255) NOT NULL,\n"
+                + "  id %2$s NOT NULL,\n"
                 + "  spec_id VARCHAR(255) NOT NULL,\n"
-                + "  version VARCHAR(255) NOT NULL,\n"
-                + "  payload %2$s NOT NULL,\n"
+                + "  created_date VARCHAR(255) NOT NULL,\n"
+                + "  payload %3$s NOT NULL,\n"
                 + "  PRIMARY KEY (id)\n"
                 + ")",
-                tableName, getPayloadType()
+                tableName, getSerialType(), getPayloadType()
             ),
             String.format("CREATE INDEX idx_%1$s_spec_id ON %1$s(spec_id)", tableName)
         )
diff --git a/server/src/main/java/io/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/io/druid/metadata/SQLMetadataSupervisorManager.java
index 64690c1844f..3ebb4a3a49d 100644
--- a/server/src/main/java/io/druid/metadata/SQLMetadataSupervisorManager.java
+++ b/server/src/main/java/io/druid/metadata/SQLMetadataSupervisorManager.java
@@ -84,16 +84,14 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
           @Override
           public Void withHandle(Handle handle) throws Exception
           {
-            final String version = new DateTime().toString();
             handle.createStatement(
                 String.format(
-                    "INSERT INTO %s (id, spec_id, version, payload) VALUES (:id, :spec_id, :version, :payload)",
+                    "INSERT INTO %s (spec_id, created_date, payload) VALUES (:spec_id, :created_date, :payload)",
                     getSupervisorsTable()
                 )
             )
-                  .bind("id", String.format("%s_%s", id, version))
                   .bind("spec_id", id)
-                  .bind("version", version)
+                  .bind("created_date", new DateTime().toString())
                   .bind("payload", jsonMapper.writeValueAsBytes(spec))
                   .execute();
 
@@ -115,7 +113,7 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
               {
                 return handle.createQuery(
                     String.format(
-                        "SELECT spec_id, version, payload FROM %1$s ORDER BY version DESC",
+                        "SELECT id, spec_id, created_date, payload FROM %1$s ORDER BY id DESC",
                         getSupervisorsTable()
                     )
                 ).map(
@@ -134,7 +132,7 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
                           );
                           return Pair.of(
                               r.getString("spec_id"),
-                              new VersionedSupervisorSpec(payload, r.getString("version"))
+                              new VersionedSupervisorSpec(payload, r.getString("created_date"))
                           );
                         }
                         catch (IOException e) {
@@ -189,8 +187,8 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
                     String.format(
                         "SELECT r.spec_id, r.payload "
                         + "FROM %1$s r "
-                        + "INNER JOIN(SELECT spec_id, max(version) as version FROM %1$s GROUP BY spec_id) latest "
-                        + "ON r.spec_id = latest.spec_id and r.version = latest.version",
+                        + "INNER JOIN(SELECT spec_id, max(id) as id FROM %1$s GROUP BY spec_id) latest "
+                        + "ON r.id = latest.id",
                         getSupervisorsTable()
                     )
                 ).map(