diff --git a/mysql-test/r/slow_log_json.result b/mysql-test/r/slow_log_json.result
new file mode 100644
index 00000000000..504777e8692
--- /dev/null
+++ b/mysql-test/r/slow_log_json.result
@@ -0,0 +1,101 @@
+SET @save_sql      = @@global.slow_query_log;
+SET @save_sqlf     = @@global.slow_query_log_file;
+SET @save_format   = @@global.slow_query_log_format;
+SET @save_lqt     = @@global.long_query_time;
+SET @save_lse     = @@global.log_slow_extra;
+# Default value is TRADITIONAL.
+SET GLOBAL slow_query_log_format = DEFAULT;
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+TRADITIONAL
+# The variable is GLOBAL only (no SESSION scope).
+SELECT @@session.slow_query_log_format;
+ERROR HY000: Variable 'slow_query_log_format' is a GLOBAL variable
+SET @@session.slow_query_log_format = JSON;
+ERROR HY000: Variable 'slow_query_log_format' is a GLOBAL variable and should be set with SET GLOBAL
+# Valid values: TRADITIONAL, JSON, and their numeric equivalents.
+SET GLOBAL slow_query_log_format = TRADITIONAL;
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+TRADITIONAL
+SET GLOBAL slow_query_log_format = JSON;
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+JSON
+SET GLOBAL slow_query_log_format = 0;
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+TRADITIONAL
+SET GLOBAL slow_query_log_format = 1;
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+JSON
+SET GLOBAL slow_query_log_format = 'TRADITIONAL';
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+TRADITIONAL
+SET GLOBAL slow_query_log_format = 'JSON';
+SELECT @@global.slow_query_log_format;
+@@global.slow_query_log_format
+JSON
+# Invalid values are rejected.
+SET GLOBAL slow_query_log_format = 'invalid';
+ERROR 42000: Variable 'slow_query_log_format' can't be set to the value of 'invalid'
+SET GLOBAL slow_query_log_format = 2;
+ERROR 42000: Variable 'slow_query_log_format' can't be set to the value of '2'
+SET GLOBAL slow_query_log_format = -1;
+ERROR 42000: Variable 'slow_query_log_format' can't be set to the value of '-1'
+SET GLOBAL slow_query_log_format = '';
+ERROR 42000: Variable 'slow_query_log_format' can't be set to the value of ''
+SET GLOBAL slow_query_log_format = NULL;
+ERROR 42000: Variable 'slow_query_log_format' can't be set to the value of 'NULL'
+# The variable is exposed via performance_schema.global_variables.
+SELECT VARIABLE_NAME, VARIABLE_VALUE FROM performance_schema.global_variables
+WHERE VARIABLE_NAME='slow_query_log_format';
+VARIABLE_NAME	VARIABLE_VALUE
+slow_query_log_format	JSON
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL slow_query_log_format = JSON;
+SET GLOBAL log_slow_extra       = OFF;
+SET GLOBAL long_query_time      = 0;
+SET GLOBAL slow_query_log_file = '.../slow_log_json.log';
+SET GLOBAL slow_query_log       = ON;
+SELECT 'json-slow-marker' AS marker;
+marker
+json-slow-marker
+SET GLOBAL slow_query_log = OFF;
+shape_ok    = 1
+header_seen = 1
+header_keys = 4
+record_seen = 0
+record_keys = 0
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL log_slow_extra       = ON;
+SET GLOBAL slow_query_log_file = '.../slow_log_json_extra.log';
+SET GLOBAL slow_query_log       = ON;
+CREATE TABLE t1 (i INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1), (2), (3);
+SELECT i FROM t1 WHERE i = 2;
+i
+2
+DROP TABLE t1;
+SET GLOBAL slow_query_log = OFF;
+shape_ok              = 1
+extra_records_matched = 0
+extra_keys_missing    = 0
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL slow_query_log_format = TRADITIONAL;
+SET GLOBAL log_slow_extra       = OFF;
+SET GLOBAL slow_query_log_file = '.../slow_log_traditional.log';
+SET GLOBAL slow_query_log       = ON;
+SELECT 'traditional-marker' AS marker;
+marker
+traditional-marker
+SET GLOBAL slow_query_log = OFF;
+traditional_header_ok  = 1
+traditional_marker_ok  = 0
+SET @@global.slow_query_log_file    = @save_sqlf;
+SET @@global.slow_query_log_format  = @save_format;
+SET @@global.long_query_time        = @save_lqt;
+SET @@global.log_slow_extra         = @save_lse;
+SET @@global.slow_query_log         = @save_sql;
diff --git a/mysql-test/t/slow_log_json.test b/mysql-test/t/slow_log_json.test
new file mode 100644
index 00000000000..125ab64d8ff
--- /dev/null
+++ b/mysql-test/t/slow_log_json.test
@@ -0,0 +1,221 @@
+--source include/no_ps_protocol.inc
+--source include/not_valgrind.inc
+--source include/not_parallel.inc
+
+# Save state.
+SET @save_sql      = @@global.slow_query_log;
+SET @save_sqlf     = @@global.slow_query_log_file;
+SET @save_format   = @@global.slow_query_log_format;
+SET @save_lqt     = @@global.long_query_time;
+SET @save_lse     = @@global.log_slow_extra;
+
+# System variable: slow_query_log_format test
+
+--echo # Default value is TRADITIONAL.
+SET GLOBAL slow_query_log_format = DEFAULT;
+SELECT @@global.slow_query_log_format;
+
+--echo # The variable is GLOBAL only (no SESSION scope).
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@session.slow_query_log_format;
+--error ER_GLOBAL_VARIABLE
+SET @@session.slow_query_log_format = JSON;
+
+--echo # Valid values: TRADITIONAL, JSON, and their numeric equivalents.
+SET GLOBAL slow_query_log_format = TRADITIONAL;
+SELECT @@global.slow_query_log_format;
+SET GLOBAL slow_query_log_format = JSON;
+SELECT @@global.slow_query_log_format;
+SET GLOBAL slow_query_log_format = 0;
+SELECT @@global.slow_query_log_format;
+SET GLOBAL slow_query_log_format = 1;
+SELECT @@global.slow_query_log_format;
+SET GLOBAL slow_query_log_format = 'TRADITIONAL';
+SELECT @@global.slow_query_log_format;
+SET GLOBAL slow_query_log_format = 'JSON';
+SELECT @@global.slow_query_log_format;
+
+--echo # Invalid values are rejected.
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL slow_query_log_format = 'invalid';
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL slow_query_log_format = 2;
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL slow_query_log_format = -1;
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL slow_query_log_format = '';
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL slow_query_log_format = NULL;
+
+--echo # The variable is exposed via performance_schema.global_variables.
+--disable_warnings
+SELECT VARIABLE_NAME, VARIABLE_VALUE FROM performance_schema.global_variables
+ WHERE VARIABLE_NAME='slow_query_log_format';
+--enable_warnings
+
+# JSON-format header and basic slow record. log_slow_extra = OFF.
+
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL slow_query_log_format = JSON;
+SET GLOBAL log_slow_extra       = OFF;
+SET GLOBAL long_query_time      = 0;
+--replace_result $MYSQL_TMP_DIR ...
+eval SET GLOBAL slow_query_log_file = '$MYSQL_TMP_DIR/slow_log_json.log';
+SET GLOBAL slow_query_log       = ON;
+
+SELECT 'json-slow-marker' AS marker;
+
+SET GLOBAL slow_query_log = OFF;
+
+let SLOW_LOG_JSON = `SELECT @@global.slow_query_log_file`;
+
+--perl
+use strict;
+use warnings;
+
+my $file = $ENV{'SLOW_LOG_JSON'} or die "slow log path not set";
+open(my $fh, '<', $file) or die "cannot open $file: $!";
+my @lines = <$fh>;
+close($fh);
+
+my @hdr_keys = qw(program version compilation_comment tcp_port);
+my @rec_keys = qw(ts user_host query_time lock_time rows_sent rows_examined
+                  id timestamp query);
+
+my $shape_ok    = 1;
+my $header_seen = 0;
+my $header_keys = 0;
+my $record_seen = 0;
+my $record_keys = 0;
+
+foreach my $line (@lines) {
+  chomp $line;
+  next if $line =~ /^\s*$/;
+  unless ($line =~ /^\{.*\}$/) {
+    print "BAD JSON SHAPE: $line\n";
+    $shape_ok = 0;
+    next;
+  }
+  if ($line =~ /"program"\s*:/) {
+    $header_seen = 1;
+    foreach my $k (@hdr_keys) {
+      $header_keys++ if $line =~ /"\Q$k\E"\s*:/;
+    }
+  }
+  if ($line =~ /"query"\s*:\s*"SELECT 'json-slow-marker' AS marker"/) {
+    $record_seen = 1;
+    foreach my $k (@rec_keys) {
+      $record_keys++ if $line =~ /"\Q$k\E"\s*:/;
+    }
+  }
+}
+
+print "shape_ok    = $shape_ok\n";
+print "header_seen = $header_seen\n";
+print "header_keys = $header_keys\n";
+print "record_seen = $record_seen\n";
+print "record_keys = $record_keys\n";
+EOF
+
+--remove_file $MYSQL_TMP_DIR/slow_log_json.log
+
+# JSON record with log_slow_extra=ON.
+
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL log_slow_extra       = ON;
+--replace_result $MYSQL_TMP_DIR ...
+eval SET GLOBAL slow_query_log_file = '$MYSQL_TMP_DIR/slow_log_json_extra.log';
+SET GLOBAL slow_query_log       = ON;
+
+CREATE TABLE t1 (i INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1), (2), (3);
+SELECT i FROM t1 WHERE i = 2;
+DROP TABLE t1;
+
+SET GLOBAL slow_query_log = OFF;
+
+let SLOW_LOG_JSON_EXTRA = `SELECT @@global.slow_query_log_file`;
+
+--perl
+use strict;
+use warnings;
+
+my $file = $ENV{'SLOW_LOG_JSON_EXTRA'} or die "slow log path not set";
+open(my $fh, '<', $file) or die "cannot open $file: $!";
+my @lines = <$fh>;
+close($fh);
+
+my @extra_keys = qw(
+  ts user_host query_time lock_time rows_sent rows_examined thread_id errno
+  killed bytes_received bytes_sent read_first read_last read_key read_next
+  read_prev read_rnd read_rnd_next sort_merge_passes sort_range_count
+  sort_rows sort_scan_count created_tmp_disk_tables created_tmp_tables
+  start end timestamp query
+);
+
+my $matched = 0;
+my $missing_total = 0;
+my $shape_ok = 1;
+
+foreach my $line (@lines) {
+  chomp $line;
+  next if $line =~ /^\s*$/;
+  unless ($line =~ /^\{.*\}$/) {
+    print "BAD JSON SHAPE: $line\n";
+    $shape_ok = 0;
+    next;
+  }
+  next unless $line =~ /"query"\s*:\s*"SELECT i FROM t1 WHERE i = 2"/;
+  $matched++;
+  foreach my $k (@extra_keys) {
+    $missing_total++ unless $line =~ /"\Q$k\E"\s*:/;
+  }
+}
+
+print "shape_ok              = $shape_ok\n";
+print "extra_records_matched = $matched\n";
+print "extra_keys_missing    = $missing_total\n";
+EOF
+
+--remove_file $MYSQL_TMP_DIR/slow_log_json_extra.log
+
+
+# Switching back to TRADITIONAL produces the legacy header
+SET GLOBAL slow_query_log       = OFF;
+SET GLOBAL slow_query_log_format = TRADITIONAL;
+SET GLOBAL log_slow_extra       = OFF;
+--replace_result $MYSQL_TMP_DIR ...
+eval SET GLOBAL slow_query_log_file = '$MYSQL_TMP_DIR/slow_log_traditional.log';
+SET GLOBAL slow_query_log       = ON;
+
+SELECT 'traditional-marker' AS marker;
+SET GLOBAL slow_query_log = OFF;
+
+let SLOW_LOG_TRAD = `SELECT @@global.slow_query_log_file`;
+
+--perl
+use strict;
+use warnings;
+
+my $file = $ENV{'SLOW_LOG_TRAD'} or die "slow log path not set";
+open(my $fh, '<', $file) or die "cannot open $file: $!";
+my $head_ok = 0;
+my $marker_ok = 0;
+while (my $line = <$fh>) {
+  $head_ok = 1   if $line =~ /Time\s+Id Command\s+Argument/;
+  $marker_ok = 1 if $line =~ /SELECT 'traditional-marker' AS marker/;
+}
+close($fh);
+print "traditional_header_ok  = $head_ok\n";
+print "traditional_marker_ok  = $marker_ok\n";
+EOF
+
+--remove_file $MYSQL_TMP_DIR/slow_log_traditional.log
+
+
+# Clean up
+SET @@global.slow_query_log_file    = @save_sqlf;
+SET @@global.slow_query_log_format  = @save_format;
+SET @@global.long_query_time        = @save_lqt;
+SET @@global.log_slow_extra         = @save_lse;
+SET @@global.slow_query_log         = @save_sql;
diff --git a/sql/log.cc b/sql/log.cc
index bcfd73a25a5..70983a8728c 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -352,6 +352,31 @@ class File_query_log {
                   ulonglong lock_utime, bool is_command, const char *sql_text,
                   size_t sql_text_len);
 
+  /**
+     Log a query to the JSON format slow log file.
+
+     @param thd                THD of the query
+     @param current_utime      Current timestamp in microseconds
+     @param query_start_utime  Command start timestamp in microseconds
+     @param user_host          The pointer to the string with user\@host info
+     @param user_host_len      Length of the user_host string
+     @param query_utime        Number of microseconds query execution took
+     @param lock_utime         Number of microseconds the query was locked
+     @param is_command         The flag which determines whether the sql_text
+                               is a query or an administrator command
+     @param sql_text           The query or administrator in textual form
+     @param sql_text_len       The length of sql_text string
+     @param aggregate          True if writing log throttle record
+
+     @return true if error, false otherwise.
+  */
+  bool write_slow_json(THD *thd, ulonglong current_utime,
+                  ulonglong query_start_utime, const char *user_host,
+                  size_t user_host_len, ulonglong query_utime,
+                  ulonglong lock_utime, bool is_command, const char *sql_text,
+                  size_t sql_text_len, bool aggregate);
+
+
  private:
   /** Type of log file. */
   const enum_log_table_type m_log_type;
@@ -553,26 +578,45 @@ bool File_query_log::open() {
     goto err;
 
   {
-    char *end;
-    const size_t len =
-        snprintf(buff, sizeof(buff),
-                 "%s, Version: %s (%s). "
+    if (opt_slow_query_log_format == 0) {
+      char *end;
+      const size_t len =
+          snprintf(buff, sizeof(buff),
+                   "%s, Version: %s (%s). "
 #if defined(_WIN32)
-                 "started with:\nTCP Port: %d, Named Pipe: %s\n",
-                 my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER,
-                 mysqld_port, mysqld_unix_port
+                   "started with:\nTCP Port: %d, Named Pipe: %s\n",
+                   my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER,
+                   mysqld_port, mysqld_unix_port
 #else
-                 "started with:\nTcp port: %d  Unix socket: %s\n",
-                 my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER,
-                 mysqld_port, mysqld_unix_port
+                   "started with:\nTcp port: %d  Unix socket: %s\n",
+                   my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER,
+                   mysqld_port, mysqld_unix_port
 #endif
-        );
-    end =
-        my_stpncpy(buff + len, "Time                 Id Command    Argument\n",
-                   sizeof(buff) - len);
-    if (my_b_write(&log_file, (uchar *)buff, (uint)(end - buff)) ||
-        flush_io_cache(&log_file))
-      goto err;
+          );
+
+      end = my_stpncpy(buff + len,
+                       "Time                 Id Command    Argument\n",
+                       sizeof(buff) - len);
+      if (my_b_write(&log_file, (uchar *)buff, (uint)(end - buff)) ||
+          flush_io_cache(&log_file))
+        goto err;
+    }
+
+    if (opt_slow_query_log_format == 1) {
+      const size_t len =
+          snprintf(buff, sizeof(buff),
+          "{\"program\":\"%s\",\"version\":\"%s\",\"compilation_comment\":\"%s\",\"tcp_port\": %d"
+#if defined(_WIN32)
+          ", \"named_pipe\":\"%s\"}\n",
+          my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER, mysqld_port, mysqld_unix_port);
+#else
+          ", \"unix_socket\":\"%s\"}\n",
+          my_progname, server_version, MYSQL_COMPILATION_COMMENT_SERVER, mysqld_port, mysqld_unix_port);
+#endif
+      if (my_b_write(&log_file, (uchar *)buff, len) ||
+          flush_io_cache(&log_file))
+        goto err;
+    }
   }
 
   log_open = true;
@@ -860,6 +904,191 @@ err:
   return true;
 }
 
+bool File_query_log::write_slow_json(
+    THD *thd, ulonglong current_utime, ulonglong query_start_utime,
+    const char *user_host, size_t user_host_len, ulonglong query_utime,
+    ulonglong lock_utime, bool is_command, const char *sql_text,
+    size_t sql_text_len, bool aggregate) {
+  std::string log_json;
+  log_json.reserve(512 + sql_text_len * 2);
+
+  bool first = true;
+
+  auto add_str = [&](const char *key, const char *val, size_t len) {
+    if (!first) log_json += ',';
+    first = false;
+    log_json += '"'; log_json += key; log_json += "\":\"";
+    for (size_t i = 0; i < len; i++) {
+      const unsigned char c = static_cast<unsigned char>(val[i]);
+      if      (c == '"')  { log_json += "\\\""; }
+      else if (c == '\\') { log_json += "\\\\"; }
+      else if (c < 0x20) {
+        char esc[8];
+        snprintf(esc, sizeof(esc), "\\u%04x", static_cast<unsigned>(c));
+        log_json += esc;
+      } else if (c < 0x80) {
+        log_json += static_cast<char>(c);
+      } else {
+        char esc[8];
+        snprintf(esc, sizeof(esc), "\\u%04x", static_cast<unsigned>(c));
+        log_json += esc;
+      }
+    }
+    log_json += '"';
+  };
+
+  auto add_num = [&](const char *key, ulonglong val) {
+    if (!first) log_json += ',';
+    first = false;
+    char buf[64];
+    snprintf(buf, sizeof(buf), "\"%s\":%llu", key, val);
+    log_json += buf;
+  };
+
+  auto add_double = [&](const char *key, double val) {
+    if (!first) log_json += ',';
+    first = false;
+    char buf[64];
+    snprintf(buf, sizeof(buf), "\"%s\":%.6f", key, val);
+    log_json += buf;
+  };
+
+  double query_time = ulonglong2double(query_utime) / 1000000.0;
+  double lock_time = ulonglong2double(lock_utime) / 1000000.0;
+
+  log_json += '{';
+
+  mysql_mutex_lock(&LOCK_log);
+  assert(is_open());
+
+  if (aggregate) {
+    add_str("type", "throttle_summary", 16);
+    add_double("query_time", query_time);
+    add_double("lock_time", lock_time);
+    add_str("extra", sql_text, sql_text_len);
+    goto write;
+  }
+
+  if (!(specialflag & SPECIAL_SHORT_LOG_FORMAT)) {
+    char my_timestamp[iso8601_size];
+
+    make_iso8601_timestamp(my_timestamp, current_utime,
+                           iso8601_sysvar_logtimestamps);
+    add_str("ts", my_timestamp, strlen(my_timestamp));
+    add_str("user_host", user_host, user_host_len);
+  }
+
+  if (!thd->copy_status_var_ptr) {
+    add_double("query_time", query_time);
+    add_double("lock_time", lock_time);
+    add_num("rows_sent", thd->get_sent_row_count());
+    add_num("rows_examined", thd->get_examined_row_count());
+    add_num("id", thd->thread_id());
+  } else {
+    char start_time_buff[iso8601_size];
+    char end_time_buff[iso8601_size];
+
+    if (query_start_utime) {
+      make_iso8601_timestamp(start_time_buff, query_start_utime,
+                             iso8601_sysvar_logtimestamps);
+      make_iso8601_timestamp(end_time_buff, query_start_utime + query_utime,
+                             iso8601_sysvar_logtimestamps);
+    } else {
+      start_time_buff[0] = '\0'; /* purecov: inspected */
+      make_iso8601_timestamp(
+          end_time_buff, current_utime,
+          iso8601_sysvar_logtimestamps); /* purecov: inspected */
+    }
+
+    add_double("query_time", query_time);
+    add_double("lock_time", lock_time);
+    add_num("rows_sent", thd->get_sent_row_count());
+    add_num("rows_examined", thd->get_examined_row_count());
+    add_num("thread_id", thd->thread_id());
+    add_num("errno", thd->is_error() ? thd->get_stmt_da()->mysql_errno() : 0);
+    add_num("killed", thd->killed);
+    add_num("bytes_received", thd->status_var.bytes_received - thd->copy_status_var_ptr->bytes_received);
+    add_num("bytes_sent", thd->status_var.bytes_sent - thd->copy_status_var_ptr->bytes_sent);
+    add_num("read_first", thd->status_var.ha_read_first_count -
+                    thd->copy_status_var_ptr->ha_read_first_count);
+    add_num("read_last", thd->status_var.ha_read_last_count -
+                    thd->copy_status_var_ptr->ha_read_last_count);
+    add_num("read_key", thd->status_var.ha_read_key_count -
+                    thd->copy_status_var_ptr->ha_read_key_count);
+    add_num("read_next", thd->status_var.ha_read_next_count -
+                    thd->copy_status_var_ptr->ha_read_next_count);
+    add_num("read_prev", thd->status_var.ha_read_prev_count -
+                    thd->copy_status_var_ptr->ha_read_prev_count);
+    add_num("read_rnd", thd->status_var.ha_read_rnd_count -
+                    thd->copy_status_var_ptr->ha_read_rnd_count);
+    add_num("read_rnd_next", thd->status_var.ha_read_rnd_next_count -
+                    thd->copy_status_var_ptr->ha_read_rnd_next_count);
+    add_num("sort_merge_passes", thd->status_var.filesort_merge_passes -
+                    thd->copy_status_var_ptr->filesort_merge_passes);
+    add_num("sort_range_count", thd->status_var.filesort_range_count -
+                    thd->copy_status_var_ptr->filesort_range_count);
+    add_num("sort_rows", thd->status_var.filesort_rows -
+                    thd->copy_status_var_ptr->filesort_rows);
+    add_num("sort_scan_count", thd->status_var.filesort_scan_count -
+                    thd->copy_status_var_ptr->filesort_scan_count);
+    add_num("created_tmp_disk_tables", thd->status_var.created_tmp_disk_tables -
+                    thd->copy_status_var_ptr->created_tmp_disk_tables);
+    add_num("created_tmp_tables", thd->status_var.created_tmp_tables -
+                    thd->copy_status_var_ptr->created_tmp_tables);
+    add_str("start", start_time_buff, strlen(start_time_buff));
+    add_str("end", end_time_buff, strlen(end_time_buff));
+  }
+
+  if (thd->db().str && strcmp(thd->db().str, db)) {  // Database changed
+    add_str("db", thd->db().str, strlen(thd->db().str));
+    my_stpcpy(db, thd->db().str);
+  }
+  if (thd->stmt_depends_on_first_successful_insert_id_in_prev_stmt) {
+    char last_insert_id_buff[32];
+    longlong10_to_str(
+        (longlong)thd->first_successful_insert_id_in_prev_stmt_for_binlog,
+        last_insert_id_buff, -10);
+    add_str("last_insert_id", last_insert_id_buff, strlen(last_insert_id_buff));
+  }
+  // Save value if we do an insert.
+  if (thd->auto_inc_intervals_in_cur_stmt_for_binlog.nb_elements() > 0) {
+    if (!(specialflag & SPECIAL_SHORT_LOG_FORMAT)) {
+      char insert_id_buff[32];
+      longlong10_to_str(
+          (longlong)thd->auto_inc_intervals_in_cur_stmt_for_binlog.minimum(),
+          insert_id_buff, -10);
+      add_str("insert_id", insert_id_buff, strlen(insert_id_buff));
+    }
+  }
+
+  /*
+    The timestamp used to only be set when the query had checked the
+    start time. Now the slow log always logs the query start time.
+    This ensures logs can be used to replicate queries accurately.
+  */
+  add_num("timestamp", query_start_utime / 1e6);
+
+  if (is_command) {
+    add_str("command", sql_text, sql_text_len);
+  } else {
+    add_str("query", sql_text, sql_text_len);
+  }
+
+write:
+  if (my_b_write(&log_file, pointer_cast<const uchar *>(log_json.c_str()), log_json.size()) ||
+      my_b_write(&log_file, pointer_cast<const uchar *>("}\n"), 2) ||
+      flush_io_cache(&log_file))
+    goto err;
+
+  mysql_mutex_unlock(&LOCK_log);
+  return false;
+
+err:
+  check_and_print_write_error();
+  mysql_mutex_unlock(&LOCK_log);
+  return true;
+}
+
 bool Log_to_csv_event_handler::log_general(
     THD *thd, ulonglong event_utime, const char *user_host,
     size_t user_host_len, my_thread_id thread_id, const char *command_type,
@@ -989,7 +1218,8 @@ err:
 bool Log_to_csv_event_handler::log_slow(
     THD *thd, ulonglong current_utime, ulonglong query_start_arg,
     const char *user_host, size_t user_host_len, ulonglong query_utime,
-    ulonglong lock_utime, bool, const char *sql_text, size_t sql_text_len) {
+    ulonglong lock_utime, bool, const char *sql_text, size_t sql_text_len,
+    bool) {
   TABLE *table = nullptr;
   bool result = true;
   bool need_close = false;
@@ -1206,7 +1436,8 @@ class Log_to_file_event_handler : public Log_event_handler {
   bool log_slow(THD *thd, ulonglong current_utime, ulonglong query_start_arg,
                 const char *user_host, size_t user_host_len,
                 ulonglong query_utime, ulonglong lock_utime, bool is_command,
-                const char *sql_text, size_t sql_text_len) override;
+                const char *sql_text, size_t sql_text_len,
+                bool aggregate) override;
 
   /**
      Wrapper around File_query_log::write_general() for general log.
@@ -1243,14 +1474,21 @@ bool Log_to_file_event_handler::log_slow(
     THD *thd, ulonglong current_utime, ulonglong query_start_utime,
     const char *user_host, size_t user_host_len, ulonglong query_utime,
     ulonglong lock_utime, bool is_command, const char *sql_text,
-    size_t sql_text_len) {
+    size_t sql_text_len, bool aggregate) {
   if (!mysql_slow_log.is_open()) return false;
 
   Silence_log_table_errors error_handler;
   thd->push_internal_handler(&error_handler);
-  const bool retval = mysql_slow_log.write_slow(
+  bool retval;
+  if (opt_slow_query_log_format == 1) {
+    retval = mysql_slow_log.write_slow_json(
+      thd, current_utime, query_start_utime, user_host, user_host_len,
+      query_utime, lock_utime, is_command, sql_text, sql_text_len, aggregate);
+  } else {
+    retval = mysql_slow_log.write_slow(
       thd, current_utime, query_start_utime, user_host, user_host_len,
       query_utime, lock_utime, is_command, sql_text, sql_text_len);
+  }
   thd->pop_internal_handler();
   return retval;
 }
@@ -1346,7 +1584,8 @@ bool Query_logger::slow_log_write(THD *thd, const char *query,
                             (thd->start_time.tv_sec * 1000000ULL) +
                                 thd->start_time.tv_usec,
                             user_host_buff, user_host_len, query_utime,
-                            lock_utime, is_command, query, query_length);
+                            lock_utime, is_command, query, query_length,
+                            aggregate);
   }
 
   mysql_rwlock_unlock(&LOCK_logger);
diff --git a/sql/log.h b/sql/log.h
index 3b82a61351b..5cedaf54462 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -175,6 +175,7 @@ class Log_event_handler {
                                is a query or an administrator command
      @param sql_text           The query or administrator in textual form
      @param sql_text_len       The length of sql_text string
+     @param aggregate          True if writing log throttle record
 
      @return true if error, false otherwise.
   */
@@ -182,7 +183,8 @@ class Log_event_handler {
                         ulonglong query_start_arg, const char *user_host,
                         size_t user_host_len, ulonglong query_utime,
                         ulonglong lock_utime, bool is_command,
-                        const char *sql_text, size_t sql_text_len) = 0;
+                        const char *sql_text, size_t sql_text_len,
+                        bool aggregate) = 0;
 
   /**
      Log command to the general log.
@@ -227,7 +229,7 @@ class Log_to_csv_event_handler : public Log_event_handler {
   bool log_slow(THD *thd, ulonglong current_utime, ulonglong query_start_arg,
                 const char *user_host, size_t user_host_len,
                 ulonglong query_utime, ulonglong lock_utime, bool is_command,
-                const char *sql_text, size_t sql_text_len) override;
+                const char *sql_text, size_t sql_text_len, bool) override;
 
   /** @see Log_event_handler::log_general(). */
   bool log_general(THD *thd, ulonglong event_utime, const char *user_host,
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 992db62ef88..5aba95a61ff 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1200,6 +1200,7 @@ ulonglong log_output_options;
 bool opt_log_queries_not_using_indexes = false;
 ulong opt_log_throttle_queries_not_using_indexes = 0;
 bool opt_log_slow_extra = false;
+ulong opt_slow_query_log_format = 0;
 bool opt_disable_networking = false, opt_skip_show_db = false;
 bool opt_skip_name_resolve = false;
 bool server_id_supplied = false;
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 7d8980599f8..18c585f5447 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -163,6 +163,7 @@ extern bool opt_bin_log;
 extern bool opt_log_replica_updates;
 extern bool opt_log_unsafe_statements;
 extern bool opt_general_log, opt_slow_log, opt_general_log_raw;
+extern ulong opt_slow_query_log_format;
 extern ulonglong log_output_options;
 extern bool opt_log_queries_not_using_indexes;
 extern ulong opt_log_throttle_queries_not_using_indexes;
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 6ada71fcc29..964a0f3229e 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -5872,6 +5872,19 @@ static Sys_var_bool Sys_slow_log_extra(
     NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_slow_log_extra),
     ON_UPDATE(nullptr));
 
+
+static const char *slow_query_log_format_names[] = {
+  "TRADITIONAL", "JSON", NullS};
+
+static Sys_var_enum Sys_slow_query_log_format(
+  "slow_query_log_format",
+  "Output format for the slow query log: TRADITIONAL or JSON",
+  GLOBAL_VAR(opt_slow_query_log_format),
+  CMD_LINE(REQUIRED_ARG),
+  slow_query_log_format_names,
+  DEFAULT(0),  // 0 = TRADITIONAL
+  NO_MUTEX_GUARD, NOT_IN_BINLOG);
+
 static bool check_not_empty_set(sys_var *, THD *, set_var *var) {
   return var->save_result.ulonglong_value == 0;
 }