ovsdb: Don't convert unchanged columns during database conversion.

Column conversion involves converting it to json and back.  These are
heavy operations and completely unnecessary if the column type didn't
change.  Most of the time schema changes only add new columns/tables
without changing existing ones at all.  Clone the column instead to
save some time.

This will also save time while destroying the original database since
we will only need to reduce reference counters on unchanged datum
objects that were cloned instead of actually freeing them.

Additionally, moving the column lookup into a separate loop, so we
don't perform an shash lookup for each column of each row.

Testing with 440 MB OVN_Southbound database shows 70% speed up of the
ovsdb_convert() function.  Execution time reduced from 15 to 4.4
seconds, 3.5 of which is a post-conversion transaction replay.  Overall
time required for the online database conversion reduced from 37 to 25
seconds.

Acked-by: Han Zhou <hzhou@ovn.org>
Signed-off-by: Ilya Maximets <i.maximets@ovn.org>

1 files changed, 48 insertions, 11 deletions

diff --git a/ovsdb/file.c b/ovsdb/file.c
index fdc289ad1..2d887e53e 100644
--- a/ovsdb/file.c
+++ b/ovsdb/file.c
@@ -270,22 +270,48 @@ ovsdb_convert_table(struct ovsdb_txn *txn,
                     const struct ovsdb_table *src_table,
                     struct ovsdb_table *dst_table)
 {
+    const struct ovsdb_column **dst_columns;
+    struct ovsdb_error *error = NULL;
     const struct ovsdb_row *src_row;
+    unsigned long *src_equal;
+    struct shash_node *node;
+    size_t n_src_columns;
+
+    n_src_columns = shash_count(&src_table->schema->columns);
+    src_equal = bitmap_allocate(n_src_columns);
+    dst_columns = xzalloc(n_src_columns * sizeof *dst_columns);
+
+    SHASH_FOR_EACH (node, &src_table->schema->columns) {
+        const struct ovsdb_column *src_column = node->data;
+
+        if (src_column->index == OVSDB_COL_UUID ||
+            src_column->index == OVSDB_COL_VERSION) {
+            continue;
+        }
+
+        const struct ovsdb_column *dst_column =
+            shash_find_data(&dst_table->schema->columns, src_column->name);
+
+        if (!dst_column) {
+            continue;
+        }
+
+        dst_columns[src_column->index] = dst_column;
+
+        if (ovsdb_type_equals(&src_column->type, &dst_column->type)) {
+            bitmap_set1(src_equal, src_column->index);
+        }
+    }
+
     HMAP_FOR_EACH (src_row, hmap_node, &src_table->rows) {
         struct ovsdb_row *dst_row = ovsdb_row_create(dst_table);
         *ovsdb_row_get_uuid_rw(dst_row) = *ovsdb_row_get_uuid(src_row);
 
-        struct shash_node *node;
         SHASH_FOR_EACH (node, &src_table->schema->columns) {
             const struct ovsdb_column *src_column = node->data;
-            if (src_column->index == OVSDB_COL_UUID ||
-                src_column->index == OVSDB_COL_VERSION) {
-                continue;
-            }
+            const struct ovsdb_column *dst_column;
 
-            const struct ovsdb_column *dst_column
-                = shash_find_data(&dst_table->schema->columns,
-                                  src_column->name);
+            dst_column = dst_columns[src_column->index];
             if (!dst_column) {
                 continue;
             }
@@ -293,19 +319,30 @@ ovsdb_convert_table(struct ovsdb_txn *txn,
             ovsdb_datum_destroy(&dst_row->fields[dst_column->index],
                                 &dst_column->type);
 
-            struct ovsdb_error *error = ovsdb_datum_convert(
+            if (bitmap_is_set(src_equal, src_column->index)) {
+                /* This column didn't change - no need to convert. */
+                ovsdb_datum_clone(&dst_row->fields[dst_column->index],
+                                  &src_row->fields[src_column->index]);
+                continue;
+            }
+
+            error = ovsdb_datum_convert(
                 &dst_row->fields[dst_column->index], &dst_column->type,
                 &src_row->fields[src_column->index], &src_column->type);
             if (error) {
                 ovsdb_datum_init_empty(&dst_row->fields[dst_column->index]);
                 ovsdb_row_destroy(dst_row);
-                return error;
+                goto exit;
             }
         }
 
         ovsdb_txn_row_insert(txn, dst_row);
     }
-    return NULL;
+
+exit:
+    free(dst_columns);
+    bitmap_free(src_equal);
+    return error;
 }
 
 /* Copies the data in 'src', converts it into the schema specified in