Merge 79ab6bb47c into 9ab60935a4

Store PubSub item creation/modification timestamps as integers instead
of "$megasec:$sec:$microsec" strings.  This can improve the performance
of certain SQL queries significantly.

Thanks to Ammonit Measurement GmbH for sponsoring this work.

sql/lite.new.sql

@@ -262,8 +262,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint REFERENCES pubsub_node(nodeid) ON DELETE CASCADE,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT UNSIGNED NOT NULL,
+  modification BIGINT UNSIGNED NOT NULL,
   payload text NOT NULL DEFAULT ''
 );
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item (itemid);

sql/lite.sql

@@ -240,8 +240,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint REFERENCES pubsub_node(nodeid) ON DELETE CASCADE,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT UNSIGNED NOT NULL,
+  modification BIGINT UNSIGNED NOT NULL,
   payload text NOT NULL DEFAULT ''
 );
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item (itemid);

sql/mssql.sql

@@ -209,8 +209,8 @@ CREATE TABLE [dbo].[pubsub_item] (
         [nodeid] [bigint] NULL,
         [itemid] [varchar] (255) NOT NULL,
         [publisher] [varchar] (250) NOT NULL,
-        [creation] [varchar] (32) NOT NULL,
-        [modification] [varchar] (32) NOT NULL,
+        [creation] [bigint] NOT NULL,
+        [modification] [bigint] NOT NULL,
         [payload] [text] NOT NULL DEFAULT ''
 ) TEXTIMAGE_ON [PRIMARY];
 

sql/mysql.new.sql

@@ -280,8 +280,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT UNSIGNED NOT NULL,
+  modification BIGINT UNSIGNED NOT NULL,
   payload mediumtext NOT NULL
 ) ENGINE=InnoDB CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item(itemid(36));

sql/mysql.sql

@@ -256,8 +256,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT UNSIGNED NOT NULL,
+  modification BIGINT UNSIGNED NOT NULL,
   payload mediumtext NOT NULL
 ) ENGINE=InnoDB CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item(itemid(36));

sql/pg.new.sql

@@ -433,8 +433,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint REFERENCES pubsub_node(nodeid) ON DELETE CASCADE,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT NOT NULL,
+  modification BIGINT NOT NULL,
   payload text NOT NULL DEFAULT ''
 );
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item USING btree (itemid);

sql/pg.sql

@@ -259,8 +259,8 @@ CREATE TABLE pubsub_item (
   nodeid bigint REFERENCES pubsub_node(nodeid) ON DELETE CASCADE,
   itemid text NOT NULL,
   publisher text NOT NULL,
-  creation varchar(32) NOT NULL,
-  modification varchar(32) NOT NULL,
+  creation BIGINT NOT NULL,
+  modification BIGINT NOT NULL,
   payload text NOT NULL DEFAULT ''
 );
 CREATE INDEX i_pubsub_item_itemid ON pubsub_item USING btree (itemid);

src/ejabberd_sql_schema.erl

@@ -394,6 +394,29 @@ get_current_version(Host, Module, Schemas) ->
             Version
     end.
 
+sqlite_table_copy(Host, SchemaInfo, Table) ->
+    ejabberd_sql:sql_transaction(Host,
+        fun() ->
+            TableName = Table#sql_table.name,
+            NewTableName = <<"new_", TableName/binary>>,
+            NewTable = Table#sql_table{name = NewTableName},
+            create_table_t(SchemaInfo, NewTable),
+            SQL2 = <<"INSERT INTO ", NewTableName/binary,
+                " SELECT * FROM ", TableName/binary>>,
+            ?INFO_MSG("Copying table ~s to ~s:~n~s~n",
+                [TableName, NewTableName, SQL2]),
+            ejabberd_sql:sql_query_t(SQL2),
+            SQL3 = <<"DROP TABLE ", TableName/binary>>,
+            ?INFO_MSG("Droping old table ~s:~n~s~n",
+                [TableName, SQL2]),
+            ejabberd_sql:sql_query_t(SQL3),
+            SQL4 = <<"ALTER TABLE ", NewTableName/binary,
+                " RENAME TO ", TableName/binary>>,
+            ?INFO_MSG("Renameing table ~s to ~s:~n~s~n",
+                [NewTableName, TableName, SQL4]),
+            ejabberd_sql:sql_query_t(SQL4)
+        end).
+
 format_type(#sql_schema_info{db_type = pgsql}, Column) ->
     case Column#sql_column.type of
         text -> <<"text">>;
@@ -697,24 +720,26 @@ format_create_table(#sql_schema_info{db_type = mysql} = SchemaInfo, Table) ->
           Table#sql_table.indices).
 
 create_table(Host, SchemaInfo, Table) ->
-    ejabberd_sql:sql_query(
-      Host,
-      fun() ->
-              SQLs = format_create_table(SchemaInfo, Table),
-              ?INFO_MSG("Creating table ~s:~n~s~n",
-                        [Table#sql_table.name, SQLs]),
-              lists:foreach(
-                fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end, SQLs),
-              case Table#sql_table.post_create of
-                  undefined ->
-                      ok;
-                  F when is_function(F, 1) ->
-                      PostSQLs = F(SchemaInfo),
-                      lists:foreach(
-                        fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end,
-                        PostSQLs)
-              end
-      end).
+    ejabberd_sql:sql_query(Host,
+        fun() ->
+            create_table_t(SchemaInfo, Table)
+        end).
+
+create_table_t(SchemaInfo, Table) ->
+    SQLs = format_create_table(SchemaInfo, Table),
+    ?INFO_MSG("Creating table ~s:~n~s~n",
+        [Table#sql_table.name, SQLs]),
+    lists:foreach(
+        fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end, SQLs),
+    case Table#sql_table.post_create of
+        undefined ->
+            ok;
+        F when is_function(F, 1) ->
+            PostSQLs = F(SchemaInfo),
+            lists:foreach(
+                fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end,
+                PostSQLs)
+    end.
 
 create_tables(Host, Module, SchemaInfo, Schema) ->
     lists:foreach(
@@ -969,6 +994,74 @@ do_update_schema(Host, Module, SchemaInfo, Schema) ->
                               ok
                       end
               end;
+         ({update_primary_key, TableName, Columns1}) ->
+             Columns =
+                 case ejabberd_sql:use_new_schema() of
+                     true ->
+                         Columns1;
+                     false ->
+                         lists:delete(
+                             <<"server_host">>, Columns1)
+                 end,
+             {value, Table} =
+                 lists:keysearch(
+                     TableName, #sql_table.name, Schema#sql_schema.tables),
+             {value, Index} =
+                 lists:keysearch(
+                     Columns, #sql_index.columns, Table#sql_table.indices),
+             Res =
+             case SchemaInfo#sql_schema_info.db_type of
+                 sqlite ->
+                     sqlite_table_copy(Host, SchemaInfo, Table);
+                 pgsql ->
+                     TableName = Table#sql_table.name,
+                     SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP CONSTRAINT ",
+                         TableName/binary,"_pkey, ",
+                         "ADD PRIMARY KEY (">>,
+                         lists:join(
+                             <<", ">>,
+                             Index#sql_index.columns),
+                         <<");">>],
+                     SQL = iolist_to_binary(SQL1),
+                     ?INFO_MSG("Update primary key ~s/~p:~n~s~n",
+                         [Table#sql_table.name,
+                             Index#sql_index.columns,
+                             SQL]),
+                     ejabberd_sql:sql_query(
+                         Host,
+                         fun(_DBType, _DBVersion) ->
+                             ejabberd_sql:sql_query_t(SQL)
+                         end);
+                 mysql ->
+                     TableName = Table#sql_table.name,
+                     SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP PRIMARY KEY, "
+                     "ADD PRIMARY KEY (">>,
+                         lists:join(
+                             <<", ">>,
+                             lists:map(
+                                 fun(Col) ->
+                                     format_mysql_index_column(Table, Col)
+                                 end, Index#sql_index.columns)),
+                         <<");">>],
+                     SQL = iolist_to_binary(SQL1),
+                     ?INFO_MSG("Update primary key ~s/~p:~n~s~n",
+                         [Table#sql_table.name,
+                             Index#sql_index.columns,
+                             SQL]),
+                     ejabberd_sql:sql_query(
+                         Host,
+                         fun(_DBType, _DBVersion) ->
+                             ejabberd_sql:sql_query_t(SQL)
+                         end)
+             end,
+             case Res of
+                 {error, Error} ->
+                     ?ERROR_MSG("Failed to update table ~s: ~p",
+                         [TableName, Error]),
+                     error(Error);
+                 _ ->
+                     ok
+             end;
          ({drop_index, TableName, Columns1}) ->
               Columns =
                   case ejabberd_sql:use_new_schema() of

src/node_flat_sql.erl

@@ -844,16 +844,16 @@ set_item(Item) ->
     P = encode_jid(JID),
     Payload = Item#pubsub_item.payload,
     XML = str:join([fxml:element_to_binary(X) || X<-Payload], <<>>),
-    SM = encode_now(M),
-    SC = encode_now(C),
+    SM = misc:now_to_usec(M),
+    SC = misc:now_to_usec(C),
     ?SQL_UPSERT_T(
        "pubsub_item",
        ["!nodeid=%(Nidx)d",
         "!itemid=%(ItemId)s",
         "publisher=%(P)s",
-        "modification=%(SM)s",
+        "modification=%(SM)d",
         "payload=%(XML)s",
-        "-creation=%(SC)s"
+        "-creation=%(SC)d"
        ]),
     ok.
 
@@ -1087,20 +1087,27 @@ rsm_page(Count, Index, Offset, Items) ->
 	     first = #rsm_first{index = Offset, data = First},
 	     last = Last}.
 
+%% Convert <<"2021-08-22T19:25:52.817368Z">> to <<"1629660352817368">>.
+-spec encode_stamp(binary()) -> binary().
 encode_stamp(Stamp) ->
     try xmpp_util:decode_timestamp(Stamp) of
 	Now ->
 	    encode_now(Now)
     catch _:{bad_timestamp, _} ->
-	    Stamp % We should return a proper error to the client instead.
+	    <<"0">> % We should return a proper error to the client instead.
     end.
+
+%% Convert <<"1629660352817368">> to <<"2021-08-22T19:25:52.817368Z">>.
+-spec decode_stamp(binary()) -> binary().
 decode_stamp(Stamp) ->
     xmpp_util:encode_timestamp(decode_now(Stamp)).
 
-encode_now({T1, T2, T3}) ->
-    <<(misc:i2l(T1, 6))/binary, ":",
-      (misc:i2l(T2, 6))/binary, ":",
-      (misc:i2l(T3, 6))/binary>>.
-decode_now(NowStr) ->
-    [MS, S, US] = binary:split(NowStr, <<":">>, [global]),
-    {binary_to_integer(MS), binary_to_integer(S), binary_to_integer(US)}.
+%% Convert {1629, 660352, 817368} to <<"1629660352817368">>.
+-spec encode_now(erlang:timestamp()) -> binary().
+encode_now(Now) ->
+    integer_to_binary(misc:now_to_usec(Now)).
+
+%% Convert <<"1629660352817368">> to {1629, 660352, 817368}.
+-spec decode_now(binary()) -> erlang:timestamp().
+decode_now(Str) ->
+    misc:usec_to_now(binary_to_integer(Str)).

src/pubsub_db_sql.erl

@@ -184,15 +184,10 @@ export(_Server) ->
                                 payload = Payload}) ->
             P = jid:encode(JID),
             XML = str:join([fxml:element_to_binary(X) || X<-Payload], <<>>),
-            SM = encode_now(M),
-            SC = encode_now(C),
+            SM = misc:now_to_usec(M),
+            SC = misc:now_to_usec(C),
             [?SQL("insert into pubsub_item(itemid,nodeid,creation,modification,publisher,payload)"
-                  " values (%(ItemId)s, %(Nidx)d, %(SC)s, %(SM)s, %(P)s, %(XML)s);")];
+                  " values (%(ItemId)s, %(Nidx)d, %(SC)d, %(SM)d, %(P)s, %(XML)s);")];
            (_Host, _R) ->
             []
         end}].
-
-encode_now({T1, T2, T3}) ->
-    <<(misc:i2l(T1, 6))/binary, ":",
-      (misc:i2l(T2, 6))/binary, ":",
-      (misc:i2l(T3, 6))/binary>>.