lib/db-ctl-base.c

   1 /*
   2  * Copyright (c) 2015 Nicira, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18
  19 #include <ctype.h>
  20 #include <getopt.h>
  21 #include <unistd.h>
  22
  23 #include "db-ctl-base.h"
  24
  25 #include "command-line.h"
  26 #include "compiler.h"
  27 #include "dirs.h"
  28 #include "dynamic-string.h"
  29 #include "fatal-signal.h"
  30 #include "hash.h"
  31 #include "json.h"
  32 #include "openvswitch/vlog.h"
  33 #include "ovsdb-data.h"
  34 #include "ovsdb-idl.h"
  35 #include "ovsdb-idl-provider.h"
  36 #include "shash.h"
  37 #include "string.h"
  38 #include "table.h"
  39 #include "util.h"
  40
  41 VLOG_DEFINE_THIS_MODULE(db_ctl_base);
  42
  43 /* The IDL we're using and the current transaction, if any.
  44  * This is for use by ctl_exit() only, to allow it to clean up.
  45  * Other code should use its context arguments. */
  46 struct ovsdb_idl *the_idl;
  47 struct ovsdb_idl_txn *the_idl_txn;
  48
  49 /* Represents all tables in the schema.  User must define 'tables'
  50  * in implementation and supply via clt_init().  The definition must end
  51  * with an all-NULL entry. */
  52 static const struct ctl_table_class *tables;
  53
  54 static struct shash all_commands = SHASH_INITIALIZER(&all_commands);
  55 static const struct ctl_table_class *get_table(const char *table_name);
  56 static void set_column(const struct ctl_table_class *,
  57                        const struct ovsdb_idl_row *, const char *,
  58                        struct ovsdb_symbol_table *);
  59
  60 \f
  61 static struct option *
  62 find_option(const char *name, struct option *options, size_t n_options)
  63 {
  64     size_t i;
  65
  66     for (i = 0; i < n_options; i++) {
  67         if (!strcmp(options[i].name, name)) {
  68             return &options[i];
  69         }
  70     }
  71     return NULL;
  72 }
  73
  74 static struct option *
  75 add_option(struct option **optionsp, size_t *n_optionsp,
  76            size_t *allocated_optionsp)
  77 {
  78     if (*n_optionsp >= *allocated_optionsp) {
  79         *optionsp = x2nrealloc(*optionsp, allocated_optionsp,
  80                                sizeof **optionsp);
  81     }
  82     return &(*optionsp)[(*n_optionsp)++];
  83 }
  84
  85 /* Converts the command arguments into format that can be parsed by
  86  * bash completion script.
  87  *
  88  * Therein, arguments will be attached with following prefixes:
  89  *
  90  *    !argument :: The argument is required
  91  *    ?argument :: The argument is optional
  92  *    *argument :: The argument may appear any number (0 or more) times
  93  *    +argument :: The argument may appear one or more times
  94  *
  95  */
  96 static void
  97 print_command_arguments(const struct ctl_command_syntax *command)
  98 {
  99     /*
 100      * The argument string is parsed in reverse.  We use a stack 'oew_stack' to
 101      * keep track of nested optionals.  Whenever a ']' is encountered, we push
 102      * a bit to 'oew_stack'.  The bit is set to 1 if the ']' is not nested.
 103      * Subsequently, we pop an entry everytime '[' is met.
 104      *
 105      * We use 'whole_word_is_optional' value to decide whether or not a ! or +
 106      * should be added on encountering a space: if the optional surrounds the
 107      * whole word then it shouldn't be, but if it is only a part of the word
 108      * (i.e. [key=]value), it should be.
 109      */
 110     uint32_t oew_stack = 0;
 111
 112     const char *arguments = command->arguments;
 113     int length = strlen(arguments);
 114     if (!length) {
 115         return;
 116     }
 117
 118     /* Output buffer, written backward from end. */
 119     char *output = xmalloc(2 * length);
 120     char *outp = output + 2 * length;
 121     *--outp = '\0';
 122
 123     bool in_repeated = false;
 124     bool whole_word_is_optional = false;
 125
 126     for (const char *inp = arguments + length; inp > arguments; ) {
 127         switch (*--inp) {
 128         case ']':
 129             oew_stack <<= 1;
 130             if (inp[1] == '\0' || inp[1] == ' ' || inp[1] == '.') {
 131                 oew_stack |= 1;
 132             }
 133             break;
 134         case '[':
 135             /* Checks if the whole word is optional, and sets the
 136              * 'whole_word_is_optional' accordingly. */
 137             if ((inp == arguments || inp[-1] == ' ') && oew_stack & 1) {
 138                 *--outp = in_repeated ? '*' : '?';
 139                 whole_word_is_optional = true;
 140             } else {
 141                 *--outp = '?';
 142                 whole_word_is_optional = false;
 143             }
 144             oew_stack >>= 1;
 145             break;
 146         case ' ':
 147             if (!whole_word_is_optional) {
 148                 *--outp = in_repeated ? '+' : '!';
 149             }
 150             *--outp = ' ';
 151             in_repeated = false;
 152             whole_word_is_optional = false;
 153             break;
 154         case '.':
 155             in_repeated = true;
 156             break;
 157         default:
 158             *--outp = *inp;
 159             break;
 160         }
 161     }
 162     if (arguments[0] != '[' && outp != output + 2 * length - 1) {
 163         *--outp = in_repeated ? '+' : '!';
 164     }
 165     printf("%s", outp);
 166     free(output);
 167 }
 168
 169 static void
 170 die_if_error(char *error)
 171 {
 172     if (error) {
 173         ctl_fatal("%s", error);
 174     }
 175 }
 176
 177 static int
 178 to_lower_and_underscores(unsigned c)
 179 {
 180     return c == '-' ? '_' : tolower(c);
 181 }
 182
 183 static unsigned int
 184 score_partial_match(const char *name, const char *s)
 185 {
 186     int score;
 187
 188     if (!strcmp(name, s)) {
 189         return UINT_MAX;
 190     }
 191     for (score = 0; ; score++, name++, s++) {
 192         if (to_lower_and_underscores(*name) != to_lower_and_underscores(*s)) {
 193             break;
 194         } else if (*name == '\0') {
 195             return UINT_MAX - 1;
 196         }
 197     }
 198     return *s == '\0' ? score : 0;
 199 }
 200
 201 static struct ovsdb_symbol *
 202 create_symbol(struct ovsdb_symbol_table *symtab, const char *id, bool *newp)
 203 {
 204     struct ovsdb_symbol *symbol;
 205
 206     if (id[0] != '@') {
 207         ctl_fatal("row id \"%s\" does not begin with \"@\"", id);
 208     }
 209
 210     if (newp) {
 211         *newp = ovsdb_symbol_table_get(symtab, id) == NULL;
 212     }
 213
 214     symbol = ovsdb_symbol_table_insert(symtab, id);
 215     if (symbol->created) {
 216         ctl_fatal("row id \"%s\" may only be specified on one --id option",
 217                   id);
 218     }
 219     symbol->created = true;
 220     return symbol;
 221 }
 222
 223 static const struct ovsdb_idl_row *
 224 get_row_by_id(struct ctl_context *ctx, const struct ctl_table_class *table,
 225               const struct ctl_row_id *id, const char *record_id)
 226 {
 227     const struct ovsdb_idl_row *referrer, *final;
 228
 229     if (!id->table) {
 230         return NULL;
 231     }
 232
 233     if (!id->name_column) {
 234         if (strcmp(record_id, ".")) {
 235             return NULL;
 236         }
 237         referrer = ovsdb_idl_first_row(ctx->idl, id->table);
 238         if (!referrer || ovsdb_idl_next_row(referrer)) {
 239             return NULL;
 240         }
 241     } else {
 242         const struct ovsdb_idl_row *row;
 243
 244         referrer = NULL;
 245         for (row = ovsdb_idl_first_row(ctx->idl, id->table);
 246              row != NULL;
 247              row = ovsdb_idl_next_row(row))
 248             {
 249                 const struct ovsdb_datum *name;
 250
 251                 name = ovsdb_idl_get(row, id->name_column,
 252                                      OVSDB_TYPE_STRING, OVSDB_TYPE_VOID);
 253                 if (name->n == 1 && !strcmp(name->keys[0].string, record_id)) {
 254                     if (referrer) {
 255                         ctl_fatal("multiple rows in %s match \"%s\"",
 256                                   table->class->name, record_id);
 257                     }
 258                     referrer = row;
 259                 }
 260             }
 261     }
 262     if (!referrer) {
 263         return NULL;
 264     }
 265
 266     final = NULL;
 267     if (id->uuid_column) {
 268         const struct ovsdb_datum *uuid;
 269
 270         ovsdb_idl_txn_verify(referrer, id->uuid_column);
 271         uuid = ovsdb_idl_get(referrer, id->uuid_column,
 272                              OVSDB_TYPE_UUID, OVSDB_TYPE_VOID);
 273         if (uuid->n == 1) {
 274             final = ovsdb_idl_get_row_for_uuid(ctx->idl, table->class,
 275                                                &uuid->keys[0].uuid);
 276         }
 277     } else {
 278         final = referrer;
 279     }
 280
 281     return final;
 282 }
 283
 284 static const struct ovsdb_idl_row *
 285 get_row(struct ctl_context *ctx,
 286         const struct ctl_table_class *table, const char *record_id,
 287         bool must_exist)
 288 {
 289     const struct ovsdb_idl_row *row;
 290     struct uuid uuid;
 291
 292     row = NULL;
 293     if (uuid_from_string(&uuid, record_id)) {
 294         row = ovsdb_idl_get_row_for_uuid(ctx->idl, table->class, &uuid);
 295     }
 296     if (!row) {
 297         int i;
 298
 299         for (i = 0; i < ARRAY_SIZE(table->row_ids); i++) {
 300             row = get_row_by_id(ctx, table, &table->row_ids[i], record_id);
 301             if (row) {
 302                 break;
 303             }
 304         }
 305     }
 306     if (must_exist && !row) {
 307         ctl_fatal("no row \"%s\" in table %s",
 308                   record_id, table->class->name);
 309     }
 310     return row;
 311 }
 312
 313 static char *
 314 get_column(const struct ctl_table_class *table, const char *column_name,
 315            const struct ovsdb_idl_column **columnp)
 316 {
 317     const struct ovsdb_idl_column *best_match = NULL;
 318     unsigned int best_score = 0;
 319     size_t i;
 320
 321     for (i = 0; i < table->class->n_columns; i++) {
 322         const struct ovsdb_idl_column *column = &table->class->columns[i];
 323         unsigned int score = score_partial_match(column->name, column_name);
 324         if (score > best_score) {
 325             best_match = column;
 326             best_score = score;
 327         } else if (score == best_score) {
 328             best_match = NULL;
 329         }
 330     }
 331
 332     *columnp = best_match;
 333     if (best_match) {
 334         return NULL;
 335     } else if (best_score) {
 336         return xasprintf("%s contains more than one column whose name "
 337                          "matches \"%s\"", table->class->name, column_name);
 338     } else {
 339         return xasprintf("%s does not contain a column whose name matches "
 340                          "\"%s\"", table->class->name, column_name);
 341     }
 342 }
 343
 344 static void
 345 pre_get_column(struct ctl_context *ctx,
 346                const struct ctl_table_class *table, const char *column_name,
 347                const struct ovsdb_idl_column **columnp)
 348 {
 349     die_if_error(get_column(table, column_name, columnp));
 350     ovsdb_idl_add_column(ctx->idl, *columnp);
 351 }
 352
 353 static const struct ctl_table_class *
 354 pre_get_table(struct ctl_context *ctx, const char *table_name)
 355 {
 356     const struct ctl_table_class *table_class;
 357     int i;
 358
 359     table_class = get_table(table_name);
 360     ovsdb_idl_add_table(ctx->idl, table_class->class);
 361
 362     for (i = 0; i < ARRAY_SIZE(table_class->row_ids); i++) {
 363         const struct ctl_row_id *id = &table_class->row_ids[i];
 364         if (id->table) {
 365             ovsdb_idl_add_table(ctx->idl, id->table);
 366         }
 367         if (id->name_column) {
 368             ovsdb_idl_add_column(ctx->idl, id->name_column);
 369         }
 370         if (id->uuid_column) {
 371             ovsdb_idl_add_column(ctx->idl, id->uuid_column);
 372         }
 373     }
 374
 375     return table_class;
 376 }
 377
 378 static char *
 379 missing_operator_error(const char *arg, const char **allowed_operators,
 380                        size_t n_allowed)
 381 {
 382     struct ds s;
 383
 384     ds_init(&s);
 385     ds_put_format(&s, "%s: argument does not end in ", arg);
 386     ds_put_format(&s, "\"%s\"", allowed_operators[0]);
 387     if (n_allowed == 2) {
 388         ds_put_format(&s, " or \"%s\"", allowed_operators[1]);
 389     } else if (n_allowed > 2) {
 390         size_t i;
 391
 392         for (i = 1; i < n_allowed - 1; i++) {
 393             ds_put_format(&s, ", \"%s\"", allowed_operators[i]);
 394         }
 395         ds_put_format(&s, ", or \"%s\"", allowed_operators[i]);
 396     }
 397     ds_put_format(&s, " followed by a value.");
 398
 399     return ds_steal_cstr(&s);
 400 }
 401
 402 /* Breaks 'arg' apart into a number of fields in the following order:
 403  *
 404  *      - The name of a column in 'table', stored into '*columnp'.  The column
 405  *        name may be abbreviated.
 406  *
 407  *      - Optionally ':' followed by a key string.  The key is stored as a
 408  *        malloc()'d string into '*keyp', or NULL if no key is present in
 409  *        'arg'.
 410  *
 411  *      - If 'valuep' is nonnull, an operator followed by a value string.  The
 412  *        allowed operators are the 'n_allowed' string in 'allowed_operators',
 413  *        or just "=" if 'n_allowed' is 0.  If 'operatorp' is nonnull, then the
 414  *        index of the operator within 'allowed_operators' is stored into
 415  *        '*operatorp'.  The value is stored as a malloc()'d string into
 416  *        '*valuep', or NULL if no value is present in 'arg'.
 417  *
 418  * On success, returns NULL.  On failure, returned a malloc()'d string error
 419  * message and stores NULL into all of the nonnull output arguments. */
 420 static char * OVS_WARN_UNUSED_RESULT
 421 parse_column_key_value(const char *arg,
 422                        const struct ctl_table_class *table,
 423                        const struct ovsdb_idl_column **columnp, char **keyp,
 424                        int *operatorp,
 425                        const char **allowed_operators, size_t n_allowed,
 426                        char **valuep)
 427 {
 428     const char *p = arg;
 429     char *column_name;
 430     char *error;
 431
 432     ovs_assert(!(operatorp && !valuep));
 433     *keyp = NULL;
 434     if (valuep) {
 435         *valuep = NULL;
 436     }
 437
 438     /* Parse column name. */
 439     error = ovsdb_token_parse(&p, &column_name);
 440     if (error) {
 441         goto error;
 442     }
 443     if (column_name[0] == '\0') {
 444         free(column_name);
 445         error = xasprintf("%s: missing column name", arg);
 446         goto error;
 447     }
 448     error = get_column(table, column_name, columnp);
 449     free(column_name);
 450     if (error) {
 451         goto error;
 452     }
 453
 454     /* Parse key string. */
 455     if (*p == ':') {
 456         p++;
 457         error = ovsdb_token_parse(&p, keyp);
 458         if (error) {
 459             goto error;
 460         }
 461     }
 462
 463     /* Parse value string. */
 464     if (valuep) {
 465         size_t best_len;
 466         size_t i;
 467         int best;
 468
 469         if (!allowed_operators) {
 470             static const char *equals = "=";
 471             allowed_operators = &equals;
 472             n_allowed = 1;
 473         }
 474
 475         best = -1;
 476         best_len = 0;
 477         for (i = 0; i < n_allowed; i++) {
 478             const char *op = allowed_operators[i];
 479             size_t op_len = strlen(op);
 480
 481             if (op_len > best_len && !strncmp(op, p, op_len) && p[op_len]) {
 482                 best_len = op_len;
 483                 best = i;
 484             }
 485         }
 486         if (best < 0) {
 487             error = missing_operator_error(arg, allowed_operators, n_allowed);
 488             goto error;
 489         }
 490
 491         if (operatorp) {
 492             *operatorp = best;
 493         }
 494         *valuep = xstrdup(p + best_len);
 495     } else {
 496         if (*p != '\0') {
 497             error = xasprintf("%s: trailing garbage \"%s\" in argument",
 498                               arg, p);
 499             goto error;
 500         }
 501     }
 502     return NULL;
 503
 504  error:
 505     *columnp = NULL;
 506     free(*keyp);
 507     *keyp = NULL;
 508     if (valuep) {
 509         free(*valuep);
 510         *valuep = NULL;
 511         if (operatorp) {
 512             *operatorp = -1;
 513         }
 514     }
 515     return error;
 516 }
 517
 518 static const struct ovsdb_idl_column *
 519 pre_parse_column_key_value(struct ctl_context *ctx,
 520                            const char *arg,
 521                            const struct ctl_table_class *table)
 522 {
 523     const struct ovsdb_idl_column *column;
 524     const char *p;
 525     char *column_name;
 526
 527     p = arg;
 528     die_if_error(ovsdb_token_parse(&p, &column_name));
 529     if (column_name[0] == '\0') {
 530         ctl_fatal("%s: missing column name", arg);
 531     }
 532
 533     pre_get_column(ctx, table, column_name, &column);
 534     free(column_name);
 535
 536     return column;
 537 }
 538
 539 static void
 540 check_mutable(const struct ovsdb_idl_row *row,
 541               const struct ovsdb_idl_column *column)
 542 {
 543     if (!ovsdb_idl_is_mutable(row, column)) {
 544         ctl_fatal("cannot modify read-only column %s in table %s",
 545                   column->name, row->table->class->name);
 546     }
 547 }
 548
 549 #define RELOPS                                  \
 550     RELOP(RELOP_EQ,     "=")                    \
 551     RELOP(RELOP_NE,     "!=")                   \
 552     RELOP(RELOP_LT,     "<")                    \
 553     RELOP(RELOP_GT,     ">")                    \
 554     RELOP(RELOP_LE,     "<=")                   \
 555     RELOP(RELOP_GE,     ">=")                   \
 556     RELOP(RELOP_SET_EQ, "{=}")                  \
 557     RELOP(RELOP_SET_NE, "{!=}")                 \
 558     RELOP(RELOP_SET_LT, "{<}")                  \
 559     RELOP(RELOP_SET_GT, "{>}")                  \
 560     RELOP(RELOP_SET_LE, "{<=}")                 \
 561     RELOP(RELOP_SET_GE, "{>=}")
 562
 563 enum relop {
 564 #define RELOP(ENUM, STRING) ENUM,
 565     RELOPS
 566 #undef RELOP
 567 };
 568
 569 static bool
 570 is_set_operator(enum relop op)
 571 {
 572     return (op == RELOP_SET_EQ || op == RELOP_SET_NE ||
 573             op == RELOP_SET_LT || op == RELOP_SET_GT ||
 574             op == RELOP_SET_LE || op == RELOP_SET_GE);
 575 }
 576
 577 static bool
 578 evaluate_relop(const struct ovsdb_datum *a, const struct ovsdb_datum *b,
 579                const struct ovsdb_type *type, enum relop op)
 580 {
 581     switch (op) {
 582     case RELOP_EQ:
 583     case RELOP_SET_EQ:
 584         return ovsdb_datum_compare_3way(a, b, type) == 0;
 585     case RELOP_NE:
 586     case RELOP_SET_NE:
 587         return ovsdb_datum_compare_3way(a, b, type) != 0;
 588     case RELOP_LT:
 589         return ovsdb_datum_compare_3way(a, b, type) < 0;
 590     case RELOP_GT:
 591         return ovsdb_datum_compare_3way(a, b, type) > 0;
 592     case RELOP_LE:
 593         return ovsdb_datum_compare_3way(a, b, type) <= 0;
 594     case RELOP_GE:
 595         return ovsdb_datum_compare_3way(a, b, type) >= 0;
 596
 597     case RELOP_SET_LT:
 598         return b->n > a->n && ovsdb_datum_includes_all(a, b, type);
 599     case RELOP_SET_GT:
 600         return a->n > b->n && ovsdb_datum_includes_all(b, a, type);
 601     case RELOP_SET_LE:
 602         return ovsdb_datum_includes_all(a, b, type);
 603     case RELOP_SET_GE:
 604         return ovsdb_datum_includes_all(b, a, type);
 605
 606     default:
 607         OVS_NOT_REACHED();
 608     }
 609 }
 610
 611 static bool
 612 is_condition_satisfied(const struct ctl_table_class *table,
 613                        const struct ovsdb_idl_row *row, const char *arg,
 614                        struct ovsdb_symbol_table *symtab)
 615 {
 616     static const char *operators[] = {
 617 #define RELOP(ENUM, STRING) STRING,
 618         RELOPS
 619 #undef RELOP
 620     };
 621
 622     const struct ovsdb_idl_column *column;
 623     const struct ovsdb_datum *have_datum;
 624     char *key_string, *value_string;
 625     struct ovsdb_type type;
 626     int operator;
 627     bool retval;
 628     char *error;
 629
 630     error = parse_column_key_value(arg, table, &column, &key_string,
 631                                    &operator, operators, ARRAY_SIZE(operators),
 632                                    &value_string);
 633     die_if_error(error);
 634     if (!value_string) {
 635         ctl_fatal("%s: missing value", arg);
 636     }
 637
 638     type = column->type;
 639     type.n_max = UINT_MAX;
 640
 641     have_datum = ovsdb_idl_read(row, column);
 642     if (key_string) {
 643         union ovsdb_atom want_key;
 644         struct ovsdb_datum b;
 645         unsigned int idx;
 646
 647         if (column->type.value.type == OVSDB_TYPE_VOID) {
 648             ctl_fatal("cannot specify key to check for non-map column %s",
 649                       column->name);
 650         }
 651
 652         die_if_error(ovsdb_atom_from_string(&want_key, &column->type.key,
 653                                             key_string, symtab));
 654
 655         type.key = type.value;
 656         type.value.type = OVSDB_TYPE_VOID;
 657         die_if_error(ovsdb_datum_from_string(&b, &type, value_string, symtab));
 658
 659         idx = ovsdb_datum_find_key(have_datum,
 660                                    &want_key, column->type.key.type);
 661         if (idx == UINT_MAX && !is_set_operator(operator)) {
 662             retval = false;
 663         } else {
 664             struct ovsdb_datum a;
 665
 666             if (idx != UINT_MAX) {
 667                 a.n = 1;
 668                 a.keys = &have_datum->values[idx];
 669                 a.values = NULL;
 670             } else {
 671                 a.n = 0;
 672                 a.keys = NULL;
 673                 a.values = NULL;
 674             }
 675
 676             retval = evaluate_relop(&a, &b, &type, operator);
 677         }
 678
 679         ovsdb_atom_destroy(&want_key, column->type.key.type);
 680         ovsdb_datum_destroy(&b, &type);
 681     } else {
 682         struct ovsdb_datum want_datum;
 683
 684         die_if_error(ovsdb_datum_from_string(&want_datum, &column->type,
 685                                              value_string, symtab));
 686         retval = evaluate_relop(have_datum, &want_datum, &type, operator);
 687         ovsdb_datum_destroy(&want_datum, &column->type);
 688     }
 689
 690     free(key_string);
 691     free(value_string);
 692
 693     return retval;
 694 }
 695
 696 static void
 697 invalidate_cache(struct ctl_context *ctx)
 698 {
 699     if (ctx->invalidate_cache) {
 700         (ctx->invalidate_cache)(ctx);
 701     }
 702 }
 703 \f
 704 static void
 705 pre_cmd_get(struct ctl_context *ctx)
 706 {
 707     const char *id = shash_find_data(&ctx->options, "--id");
 708     const char *table_name = ctx->argv[1];
 709     const struct ctl_table_class *table;
 710     int i;
 711
 712     /* Using "get" without --id or a column name could possibly make sense.
 713      * Maybe, for example, a *ctl command run wants to assert that a row
 714      * exists.  But it is unlikely that an interactive user would want to do
 715      * that, so issue a warning if we're running on a terminal. */
 716     if (!id && ctx->argc <= 3 && isatty(STDOUT_FILENO)) {
 717         VLOG_WARN("\"get\" command without row arguments or \"--id\" is "
 718                   "possibly erroneous");
 719     }
 720
 721     table = pre_get_table(ctx, table_name);
 722     for (i = 3; i < ctx->argc; i++) {
 723         if (!strcasecmp(ctx->argv[i], "_uuid")
 724             || !strcasecmp(ctx->argv[i], "-uuid")) {
 725             continue;
 726         }
 727
 728         pre_parse_column_key_value(ctx, ctx->argv[i], table);
 729     }
 730 }
 731
 732 static void
 733 cmd_get(struct ctl_context *ctx)
 734 {
 735     const char *id = shash_find_data(&ctx->options, "--id");
 736     bool must_exist = !shash_find(&ctx->options, "--if-exists");
 737     const char *table_name = ctx->argv[1];
 738     const char *record_id = ctx->argv[2];
 739     const struct ctl_table_class *table;
 740     const struct ovsdb_idl_row *row;
 741     struct ds *out = &ctx->output;
 742     int i;
 743
 744     if (id && !must_exist) {
 745         ctl_fatal("--if-exists and --id may not be specified together");
 746     }
 747
 748     table = get_table(table_name);
 749     row = get_row(ctx, table, record_id, must_exist);
 750     if (!row) {
 751         return;
 752     }
 753
 754     if (id) {
 755         struct ovsdb_symbol *symbol;
 756         bool new;
 757
 758         symbol = create_symbol(ctx->symtab, id, &new);
 759         if (!new) {
 760             ctl_fatal("row id \"%s\" specified on \"get\" command was used "
 761                       "before it was defined", id);
 762         }
 763         symbol->uuid = row->uuid;
 764
 765         /* This symbol refers to a row that already exists, so disable warnings
 766          * about it being unreferenced. */
 767         symbol->strong_ref = true;
 768     }
 769     for (i = 3; i < ctx->argc; i++) {
 770         const struct ovsdb_idl_column *column;
 771         const struct ovsdb_datum *datum;
 772         char *key_string;
 773
 774         /* Special case for obtaining the UUID of a row.  We can't just do this
 775          * through parse_column_key_value() below since it returns a "struct
 776          * ovsdb_idl_column" and the UUID column doesn't have one. */
 777         if (!strcasecmp(ctx->argv[i], "_uuid")
 778             || !strcasecmp(ctx->argv[i], "-uuid")) {
 779             ds_put_format(out, UUID_FMT"\n", UUID_ARGS(&row->uuid));
 780             continue;
 781         }
 782
 783         die_if_error(parse_column_key_value(ctx->argv[i], table,
 784                                             &column, &key_string,
 785                                             NULL, NULL, 0, NULL));
 786
 787         ovsdb_idl_txn_verify(row, column);
 788         datum = ovsdb_idl_read(row, column);
 789         if (key_string) {
 790             union ovsdb_atom key;
 791             unsigned int idx;
 792
 793             if (column->type.value.type == OVSDB_TYPE_VOID) {
 794                 ctl_fatal("cannot specify key to get for non-map column %s",
 795                           column->name);
 796             }
 797
 798             die_if_error(ovsdb_atom_from_string(&key,
 799                                                 &column->type.key,
 800                                                 key_string, ctx->symtab));
 801
 802             idx = ovsdb_datum_find_key(datum, &key,
 803                                        column->type.key.type);
 804             if (idx == UINT_MAX) {
 805                 if (must_exist) {
 806                     ctl_fatal("no key \"%s\" in %s record \"%s\" column %s",
 807                               key_string, table->class->name, record_id,
 808                               column->name);
 809                 }
 810             } else {
 811                 ovsdb_atom_to_string(&datum->values[idx],
 812                                      column->type.value.type, out);
 813             }
 814             ovsdb_atom_destroy(&key, column->type.key.type);
 815         } else {
 816             ovsdb_datum_to_string(datum, &column->type, out);
 817         }
 818         ds_put_char(out, '\n');
 819
 820         free(key_string);
 821     }
 822 }
 823
 824 static void
 825 parse_column_names(const char *column_names,
 826                    const struct ctl_table_class *table,
 827                    const struct ovsdb_idl_column ***columnsp,
 828                    size_t *n_columnsp)
 829 {
 830     const struct ovsdb_idl_column **columns;
 831     size_t n_columns;
 832
 833     if (!column_names) {
 834         size_t i;
 835
 836         n_columns = table->class->n_columns + 1;
 837         columns = xmalloc(n_columns * sizeof *columns);
 838         columns[0] = NULL;
 839         for (i = 0; i < table->class->n_columns; i++) {
 840             columns[i + 1] = &table->class->columns[i];
 841         }
 842     } else {
 843         char *s = xstrdup(column_names);
 844         size_t allocated_columns;
 845         char *save_ptr = NULL;
 846         char *column_name;
 847
 848         columns = NULL;
 849         allocated_columns = n_columns = 0;
 850         for (column_name = strtok_r(s, ", ", &save_ptr); column_name;
 851              column_name = strtok_r(NULL, ", ", &save_ptr)) {
 852             const struct ovsdb_idl_column *column;
 853
 854             if (!strcasecmp(column_name, "_uuid")) {
 855                 column = NULL;
 856             } else {
 857                 die_if_error(get_column(table, column_name, &column));
 858             }
 859             if (n_columns >= allocated_columns) {
 860                 columns = x2nrealloc(columns, &allocated_columns,
 861                                      sizeof *columns);
 862             }
 863             columns[n_columns++] = column;
 864         }
 865         free(s);
 866
 867         if (!n_columns) {
 868             ctl_fatal("must specify at least one column name");
 869         }
 870     }
 871     *columnsp = columns;
 872     *n_columnsp = n_columns;
 873 }
 874
 875 static void
 876 pre_list_columns(struct ctl_context *ctx,
 877                  const struct ctl_table_class *table,
 878                  const char *column_names)
 879 {
 880     const struct ovsdb_idl_column **columns;
 881     size_t n_columns;
 882     size_t i;
 883
 884     parse_column_names(column_names, table, &columns, &n_columns);
 885     for (i = 0; i < n_columns; i++) {
 886         if (columns[i]) {
 887             ovsdb_idl_add_column(ctx->idl, columns[i]);
 888         }
 889     }
 890     free(columns);
 891 }
 892
 893 static void
 894 pre_cmd_list(struct ctl_context *ctx)
 895 {
 896     const char *column_names = shash_find_data(&ctx->options, "--columns");
 897     const char *table_name = ctx->argv[1];
 898     const struct ctl_table_class *table;
 899
 900     table = pre_get_table(ctx, table_name);
 901     pre_list_columns(ctx, table, column_names);
 902 }
 903
 904 static struct table *
 905 list_make_table(const struct ovsdb_idl_column **columns, size_t n_columns)
 906 {
 907     struct table *out;
 908     size_t i;
 909
 910     out = xmalloc(sizeof *out);
 911     table_init(out);
 912
 913     for (i = 0; i < n_columns; i++) {
 914         const struct ovsdb_idl_column *column = columns[i];
 915         const char *column_name = column ? column->name : "_uuid";
 916
 917         table_add_column(out, "%s", column_name);
 918     }
 919
 920     return out;
 921 }
 922
 923 static void
 924 list_record(const struct ovsdb_idl_row *row,
 925             const struct ovsdb_idl_column **columns, size_t n_columns,
 926             struct table *out)
 927 {
 928     size_t i;
 929
 930     if (!row) {
 931         return;
 932     }
 933
 934     table_add_row(out);
 935     for (i = 0; i < n_columns; i++) {
 936         const struct ovsdb_idl_column *column = columns[i];
 937         struct cell *cell = table_add_cell(out);
 938
 939         if (!column) {
 940             struct ovsdb_datum datum;
 941             union ovsdb_atom atom;
 942
 943             atom.uuid = row->uuid;
 944
 945             datum.keys = &atom;
 946             datum.values = NULL;
 947             datum.n = 1;
 948
 949             cell->json = ovsdb_datum_to_json(&datum, &ovsdb_type_uuid);
 950             cell->type = &ovsdb_type_uuid;
 951         } else {
 952             const struct ovsdb_datum *datum = ovsdb_idl_read(row, column);
 953
 954             cell->json = ovsdb_datum_to_json(datum, &column->type);
 955             cell->type = &column->type;
 956         }
 957     }
 958 }
 959
 960 static void
 961 cmd_list(struct ctl_context *ctx)
 962 {
 963     const char *column_names = shash_find_data(&ctx->options, "--columns");
 964     bool must_exist = !shash_find(&ctx->options, "--if-exists");
 965     const struct ovsdb_idl_column **columns;
 966     const char *table_name = ctx->argv[1];
 967     const struct ctl_table_class *table;
 968     struct table *out;
 969     size_t n_columns;
 970     int i;
 971
 972     table = get_table(table_name);
 973     parse_column_names(column_names, table, &columns, &n_columns);
 974     out = ctx->table = list_make_table(columns, n_columns);
 975     if (ctx->argc > 2) {
 976         for (i = 2; i < ctx->argc; i++) {
 977             list_record(get_row(ctx, table, ctx->argv[i], must_exist),
 978                         columns, n_columns, out);
 979         }
 980     } else {
 981         const struct ovsdb_idl_row *row;
 982
 983         for (row = ovsdb_idl_first_row(ctx->idl, table->class); row != NULL;
 984              row = ovsdb_idl_next_row(row)) {
 985             list_record(row, columns, n_columns, out);
 986         }
 987     }
 988     free(columns);
 989 }
 990
 991 static void
 992 pre_cmd_find(struct ctl_context *ctx)
 993 {
 994     const char *column_names = shash_find_data(&ctx->options, "--columns");
 995     const char *table_name = ctx->argv[1];
 996     const struct ctl_table_class *table;
 997     int i;
 998
 999     table = pre_get_table(ctx, table_name);
1000     pre_list_columns(ctx, table, column_names);
1001     for (i = 2; i < ctx->argc; i++) {
1002         pre_parse_column_key_value(ctx, ctx->argv[i], table);
1003     }
1004 }
1005
1006 static void
1007 cmd_find(struct ctl_context *ctx)
1008 {
1009     const char *column_names = shash_find_data(&ctx->options, "--columns");
1010     const struct ovsdb_idl_column **columns;
1011     const char *table_name = ctx->argv[1];
1012     const struct ctl_table_class *table;
1013     const struct ovsdb_idl_row *row;
1014     struct table *out;
1015     size_t n_columns;
1016
1017     table = get_table(table_name);
1018     parse_column_names(column_names, table, &columns, &n_columns);
1019     out = ctx->table = list_make_table(columns, n_columns);
1020     for (row = ovsdb_idl_first_row(ctx->idl, table->class); row;
1021          row = ovsdb_idl_next_row(row)) {
1022         int i;
1023
1024         for (i = 2; i < ctx->argc; i++) {
1025             if (!is_condition_satisfied(table, row, ctx->argv[i],
1026                                         ctx->symtab)) {
1027                 goto next_row;
1028             }
1029         }
1030         list_record(row, columns, n_columns, out);
1031
1032     next_row: ;
1033     }
1034     free(columns);
1035 }
1036
1037 static void
1038 pre_cmd_set(struct ctl_context *ctx)
1039 {
1040     const char *table_name = ctx->argv[1];
1041     const struct ctl_table_class *table;
1042     int i;
1043
1044     table = pre_get_table(ctx, table_name);
1045     for (i = 3; i < ctx->argc; i++) {
1046         pre_parse_column_key_value(ctx, ctx->argv[i], table);
1047     }
1048 }
1049
1050 static void
1051 cmd_set(struct ctl_context *ctx)
1052 {
1053     bool must_exist = !shash_find(&ctx->options, "--if-exists");
1054     const char *table_name = ctx->argv[1];
1055     const char *record_id = ctx->argv[2];
1056     const struct ctl_table_class*table;
1057     const struct ovsdb_idl_row *row;
1058     int i;
1059
1060     table = get_table(table_name);
1061     row = get_row(ctx, table, record_id, must_exist);
1062     if (!row) {
1063         return;
1064     }
1065
1066     for (i = 3; i < ctx->argc; i++) {
1067         set_column(table, row, ctx->argv[i], ctx->symtab);
1068     }
1069
1070     invalidate_cache(ctx);
1071 }
1072
1073 static void
1074 pre_cmd_add(struct ctl_context *ctx)
1075 {
1076     const char *table_name = ctx->argv[1];
1077     const char *column_name = ctx->argv[3];
1078     const struct ctl_table_class *table;
1079     const struct ovsdb_idl_column *column;
1080
1081     table = pre_get_table(ctx, table_name);
1082     pre_get_column(ctx, table, column_name, &column);
1083 }
1084
1085 static void
1086 cmd_add(struct ctl_context *ctx)
1087 {
1088     bool must_exist = !shash_find(&ctx->options, "--if-exists");
1089     const char *table_name = ctx->argv[1];
1090     const char *record_id = ctx->argv[2];
1091     const char *column_name = ctx->argv[3];
1092     const struct ctl_table_class *table;
1093     const struct ovsdb_idl_column *column;
1094     const struct ovsdb_idl_row *row;
1095     const struct ovsdb_type *type;
1096     struct ovsdb_datum old;
1097     int i;
1098
1099     table = get_table(table_name);
1100     die_if_error(get_column(table, column_name, &column));
1101     row = get_row(ctx, table, record_id, must_exist);
1102     if (!row) {
1103         return;
1104     }
1105     check_mutable(row, column);
1106
1107     type = &column->type;
1108     ovsdb_datum_clone(&old, ovsdb_idl_read(row, column), &column->type);
1109     for (i = 4; i < ctx->argc; i++) {
1110         struct ovsdb_type add_type;
1111         struct ovsdb_datum add;
1112
1113         add_type = *type;
1114         add_type.n_min = 1;
1115         add_type.n_max = UINT_MAX;
1116         die_if_error(ovsdb_datum_from_string(&add, &add_type, ctx->argv[i],
1117                                              ctx->symtab));
1118         ovsdb_datum_union(&old, &add, type, false);
1119         ovsdb_datum_destroy(&add, type);
1120     }
1121     if (old.n > type->n_max) {
1122         ctl_fatal("\"add\" operation would put %u %s in column %s of "
1123                   "table %s but the maximum number is %u",
1124                   old.n,
1125                   type->value.type == OVSDB_TYPE_VOID ? "values" : "pairs",
1126                   column->name, table->class->name, type->n_max);
1127     }
1128     ovsdb_idl_txn_verify(row, column);
1129     ovsdb_idl_txn_write(row, column, &old);
1130
1131     invalidate_cache(ctx);
1132 }
1133
1134 static void
1135 pre_cmd_remove(struct ctl_context *ctx)
1136 {
1137     const char *table_name = ctx->argv[1];
1138     const char *column_name = ctx->argv[3];
1139     const struct ctl_table_class *table;
1140     const struct ovsdb_idl_column *column;
1141
1142     table = pre_get_table(ctx, table_name);
1143     pre_get_column(ctx, table, column_name, &column);
1144 }
1145
1146 static void
1147 cmd_remove(struct ctl_context *ctx)
1148 {
1149     bool must_exist = !shash_find(&ctx->options, "--if-exists");
1150     const char *table_name = ctx->argv[1];
1151     const char *record_id = ctx->argv[2];
1152     const char *column_name = ctx->argv[3];
1153     const struct ctl_table_class *table;
1154     const struct ovsdb_idl_column *column;
1155     const struct ovsdb_idl_row *row;
1156     const struct ovsdb_type *type;
1157     struct ovsdb_datum old;
1158     int i;
1159
1160     table = get_table(table_name);
1161     die_if_error(get_column(table, column_name, &column));
1162     row = get_row(ctx, table, record_id, must_exist);
1163     if (!row) {
1164         return;
1165     }
1166     check_mutable(row, column);
1167
1168     type = &column->type;
1169     ovsdb_datum_clone(&old, ovsdb_idl_read(row, column), &column->type);
1170     for (i = 4; i < ctx->argc; i++) {
1171         struct ovsdb_type rm_type;
1172         struct ovsdb_datum rm;
1173         char *error;
1174
1175         rm_type = *type;
1176         rm_type.n_min = 1;
1177         rm_type.n_max = UINT_MAX;
1178         error = ovsdb_datum_from_string(&rm, &rm_type,
1179                                         ctx->argv[i], ctx->symtab);
1180
1181         if (error) {
1182             if (ovsdb_type_is_map(&rm_type)) {
1183                 rm_type.value.type = OVSDB_TYPE_VOID;
1184                 free(error);
1185                 die_if_error(ovsdb_datum_from_string(
1186                                                      &rm, &rm_type, ctx->argv[i], ctx->symtab));
1187             } else {
1188                 ctl_fatal("%s", error);
1189             }
1190         }
1191         ovsdb_datum_subtract(&old, type, &rm, &rm_type);
1192         ovsdb_datum_destroy(&rm, &rm_type);
1193     }
1194     if (old.n < type->n_min) {
1195         ctl_fatal("\"remove\" operation would put %u %s in column %s of "
1196                   "table %s but the minimum number is %u",
1197                   old.n,
1198                   type->value.type == OVSDB_TYPE_VOID ? "values" : "pairs",
1199                   column->name, table->class->name, type->n_min);
1200     }
1201     ovsdb_idl_txn_verify(row, column);
1202     ovsdb_idl_txn_write(row, column, &old);
1203
1204     invalidate_cache(ctx);
1205 }
1206
1207 static void
1208 pre_cmd_clear(struct ctl_context *ctx)
1209 {
1210     const char *table_name = ctx->argv[1];
1211     const struct ctl_table_class *table;
1212     int i;
1213
1214     table = pre_get_table(ctx, table_name);
1215     for (i = 3; i < ctx->argc; i++) {
1216         const struct ovsdb_idl_column *column;
1217
1218         pre_get_column(ctx, table, ctx->argv[i], &column);
1219     }
1220 }
1221
1222 static void
1223 cmd_clear(struct ctl_context *ctx)
1224 {
1225     bool must_exist = !shash_find(&ctx->options, "--if-exists");
1226     const char *table_name = ctx->argv[1];
1227     const char *record_id = ctx->argv[2];
1228     const struct ctl_table_class *table;
1229     const struct ovsdb_idl_row *row;
1230     int i;
1231
1232     table = get_table(table_name);
1233     row = get_row(ctx, table, record_id, must_exist);
1234     if (!row) {
1235         return;
1236     }
1237
1238     for (i = 3; i < ctx->argc; i++) {
1239         const struct ovsdb_idl_column *column;
1240         const struct ovsdb_type *type;
1241         struct ovsdb_datum datum;
1242
1243         die_if_error(get_column(table, ctx->argv[i], &column));
1244         check_mutable(row, column);
1245
1246         type = &column->type;
1247         if (type->n_min > 0) {
1248             ctl_fatal("\"clear\" operation cannot be applied to column %s "
1249                       "of table %s, which is not allowed to be empty",
1250                       column->name, table->class->name);
1251         }
1252
1253         ovsdb_datum_init_empty(&datum);
1254         ovsdb_idl_txn_write(row, column, &datum);
1255     }
1256
1257     invalidate_cache(ctx);
1258 }
1259
1260 static void
1261 pre_create(struct ctl_context *ctx)
1262 {
1263     const char *id = shash_find_data(&ctx->options, "--id");
1264     const char *table_name = ctx->argv[1];
1265     const struct ctl_table_class *table;
1266
1267     table = get_table(table_name);
1268     if (!id && !table->class->is_root) {
1269         VLOG_WARN("applying \"create\" command to table %s without --id "
1270                   "option will have no effect", table->class->name);
1271     }
1272 }
1273
1274 static void
1275 cmd_create(struct ctl_context *ctx)
1276 {
1277     const char *id = shash_find_data(&ctx->options, "--id");
1278     const char *table_name = ctx->argv[1];
1279     const struct ctl_table_class *table = get_table(table_name);
1280     const struct ovsdb_idl_row *row;
1281     const struct uuid *uuid;
1282     int i;
1283
1284     if (id) {
1285         struct ovsdb_symbol *symbol = create_symbol(ctx->symtab, id, NULL);
1286         if (table->class->is_root) {
1287             /* This table is in the root set, meaning that rows created in it
1288              * won't disappear even if they are unreferenced, so disable
1289              * warnings about that by pretending that there is a reference. */
1290             symbol->strong_ref = true;
1291         }
1292         uuid = &symbol->uuid;
1293     } else {
1294         uuid = NULL;
1295     }
1296
1297     row = ovsdb_idl_txn_insert(ctx->txn, table->class, uuid);
1298     for (i = 2; i < ctx->argc; i++) {
1299         set_column(table, row, ctx->argv[i], ctx->symtab);
1300     }
1301     ds_put_format(&ctx->output, UUID_FMT, UUID_ARGS(&row->uuid));
1302 }
1303
1304 /* This function may be used as the 'postprocess' function for commands that
1305  * insert new rows into the database.  It expects that the command's 'run'
1306  * function prints the UUID reported by ovsdb_idl_txn_insert() as the command's
1307  * sole output.  It replaces that output by the row's permanent UUID assigned
1308  * by the database server and appends a new-line.
1309  *
1310  * Currently we use this only for "create", because the higher-level commands
1311  * are supposed to be independent of the actual structure of the vswitch
1312  * configuration. */
1313 static void
1314 post_create(struct ctl_context *ctx)
1315 {
1316     const struct uuid *real;
1317     struct uuid dummy;
1318
1319     if (!uuid_from_string(&dummy, ds_cstr(&ctx->output))) {
1320         OVS_NOT_REACHED();
1321     }
1322     real = ovsdb_idl_txn_get_insert_uuid(ctx->txn, &dummy);
1323     if (real) {
1324         ds_clear(&ctx->output);
1325         ds_put_format(&ctx->output, UUID_FMT, UUID_ARGS(real));
1326     }
1327     ds_put_char(&ctx->output, '\n');
1328 }
1329
1330 static void
1331 pre_cmd_destroy(struct ctl_context *ctx)
1332 {
1333     const char *table_name = ctx->argv[1];
1334
1335     pre_get_table(ctx, table_name);
1336 }
1337
1338 static void
1339 cmd_destroy(struct ctl_context *ctx)
1340 {
1341     bool must_exist = !shash_find(&ctx->options, "--if-exists");
1342     bool delete_all = shash_find(&ctx->options, "--all");
1343     const char *table_name = ctx->argv[1];
1344     const struct ctl_table_class *table;
1345     int i;
1346
1347     table = get_table(table_name);
1348
1349     if (delete_all && ctx->argc > 2) {
1350         ctl_fatal("--all and records argument should not be specified together");
1351     }
1352
1353     if (delete_all && !must_exist) {
1354         ctl_fatal("--all and --if-exists should not be specified together");
1355     }
1356
1357     if (delete_all) {
1358         const struct ovsdb_idl_row *row;
1359         const struct ovsdb_idl_row *next_row;
1360
1361         for (row = ovsdb_idl_first_row(ctx->idl, table->class);
1362              row;) {
1363             next_row = ovsdb_idl_next_row(row);
1364             ovsdb_idl_txn_delete(row);
1365             row = next_row;
1366         }
1367     } else {
1368         for (i = 2; i < ctx->argc; i++) {
1369             const struct ovsdb_idl_row *row;
1370
1371             row = get_row(ctx, table, ctx->argv[i], must_exist);
1372             if (row) {
1373                 ovsdb_idl_txn_delete(row);
1374             }
1375         }
1376     }
1377     invalidate_cache(ctx);
1378 }
1379
1380 static void
1381 pre_cmd_wait_until(struct ctl_context *ctx)
1382 {
1383     const char *table_name = ctx->argv[1];
1384     const struct ctl_table_class *table;
1385     int i;
1386
1387     table = pre_get_table(ctx, table_name);
1388
1389     for (i = 3; i < ctx->argc; i++) {
1390         pre_parse_column_key_value(ctx, ctx->argv[i], table);
1391     }
1392 }
1393
1394 static void
1395 cmd_wait_until(struct ctl_context *ctx)
1396 {
1397     const char *table_name = ctx->argv[1];
1398     const char *record_id = ctx->argv[2];
1399     const struct ctl_table_class *table;
1400     const struct ovsdb_idl_row *row;
1401     int i;
1402
1403     table = get_table(table_name);
1404
1405     row = get_row(ctx, table, record_id, false);
1406     if (!row) {
1407         ctx->try_again = true;
1408         return;
1409     }
1410
1411     for (i = 3; i < ctx->argc; i++) {
1412         if (!is_condition_satisfied(table, row, ctx->argv[i], ctx->symtab)) {
1413             ctx->try_again = true;
1414             return;
1415         }
1416     }
1417 }
1418
1419 /* Parses one command. */
1420 static void
1421 parse_command(int argc, char *argv[], struct shash *local_options,
1422               struct ctl_command *command)
1423 {
1424     const struct ctl_command_syntax *p;
1425     struct shash_node *node;
1426     int n_arg;
1427     int i;
1428
1429     shash_init(&command->options);
1430     shash_swap(local_options, &command->options);
1431     for (i = 0; i < argc; i++) {
1432         const char *option = argv[i];
1433         const char *equals;
1434         char *key, *value;
1435
1436         if (option[0] != '-') {
1437             break;
1438         }
1439
1440         equals = strchr(option, '=');
1441         if (equals) {
1442             key = xmemdup0(option, equals - option);
1443             value = xstrdup(equals + 1);
1444         } else {
1445             key = xstrdup(option);
1446             value = NULL;
1447         }
1448
1449         if (shash_find(&command->options, key)) {
1450             ctl_fatal("'%s' option specified multiple times", argv[i]);
1451         }
1452         shash_add_nocopy(&command->options, key, value);
1453     }
1454     if (i == argc) {
1455         ctl_fatal("missing command name (use --help for help)");
1456     }
1457
1458     p = shash_find_data(&all_commands, argv[i]);
1459     if (!p) {
1460         ctl_fatal("unknown command '%s'; use --help for help", argv[i]);
1461     }
1462
1463     SHASH_FOR_EACH (node, &command->options) {
1464         const char *s = strstr(p->options, node->name);
1465         int end = s ? s[strlen(node->name)] : EOF;
1466
1467         if (end != '=' && end != ',' && end != ' ' && end != '\0') {
1468             ctl_fatal("'%s' command has no '%s' option",
1469                       argv[i], node->name);
1470         }
1471         if ((end == '=') != (node->data != NULL)) {
1472             if (end == '=') {
1473                 ctl_fatal("missing argument to '%s' option on '%s' "
1474                           "command", node->name, argv[i]);
1475             } else {
1476                 ctl_fatal("'%s' option on '%s' does not accept an "
1477                           "argument", node->name, argv[i]);
1478             }
1479         }
1480     }
1481
1482     n_arg = argc - i - 1;
1483     if (n_arg < p->min_args) {
1484         ctl_fatal("'%s' command requires at least %d arguments",
1485                   p->name, p->min_args);
1486     } else if (n_arg > p->max_args) {
1487         int j;
1488
1489         for (j = i + 1; j < argc; j++) {
1490             if (argv[j][0] == '-') {
1491                 ctl_fatal("'%s' command takes at most %d arguments "
1492                           "(note that options must precede command "
1493                           "names and follow a \"--\" argument)",
1494                           p->name, p->max_args);
1495             }
1496         }
1497
1498         ctl_fatal("'%s' command takes at most %d arguments",
1499                   p->name, p->max_args);
1500     }
1501
1502     command->syntax = p;
1503     command->argc = n_arg + 1;
1504     command->argv = &argv[i];
1505 }
1506
1507 static void
1508 pre_cmd_show(struct ctl_context *ctx)
1509 {
1510     struct cmd_show_table *show;
1511
1512     for (show = cmd_show_tables; show->table; show++) {
1513         size_t i;
1514
1515         ovsdb_idl_add_table(ctx->idl, show->table);
1516         if (show->name_column) {
1517             ovsdb_idl_add_column(ctx->idl, show->name_column);
1518         }
1519         for (i = 0; i < ARRAY_SIZE(show->columns); i++) {
1520             const struct ovsdb_idl_column *column = show->columns[i];
1521             if (column) {
1522                 ovsdb_idl_add_column(ctx->idl, column);
1523             }
1524         }
1525     }
1526 }
1527
1528 static struct cmd_show_table *
1529 cmd_show_find_table_by_row(const struct ovsdb_idl_row *row)
1530 {
1531     struct cmd_show_table *show;
1532
1533     for (show = cmd_show_tables; show->table; show++) {
1534         if (show->table == row->table->class) {
1535             return show;
1536         }
1537     }
1538     return NULL;
1539 }
1540
1541 static struct cmd_show_table *
1542 cmd_show_find_table_by_name(const char *name)
1543 {
1544     struct cmd_show_table *show;
1545
1546     for (show = cmd_show_tables; show->table; show++) {
1547         if (!strcmp(show->table->name, name)) {
1548             return show;
1549         }
1550     }
1551     return NULL;
1552 }
1553
1554 static void
1555 cmd_show_row(struct ctl_context *ctx, const struct ovsdb_idl_row *row,
1556              int level)
1557 {
1558     struct cmd_show_table *show = cmd_show_find_table_by_row(row);
1559     size_t i;
1560
1561     ds_put_char_multiple(&ctx->output, ' ', level * 4);
1562     if (show && show->name_column) {
1563         const struct ovsdb_datum *datum;
1564
1565         ds_put_format(&ctx->output, "%s ", show->table->name);
1566         datum = ovsdb_idl_read(row, show->name_column);
1567         ovsdb_datum_to_string(datum, &show->name_column->type, &ctx->output);
1568     } else {
1569         ds_put_format(&ctx->output, UUID_FMT, UUID_ARGS(&row->uuid));
1570     }
1571     ds_put_char(&ctx->output, '\n');
1572
1573     if (!show || show->recurse) {
1574         return;
1575     }
1576
1577     show->recurse = true;
1578     for (i = 0; i < ARRAY_SIZE(show->columns); i++) {
1579         const struct ovsdb_idl_column *column = show->columns[i];
1580         const struct ovsdb_datum *datum;
1581
1582         if (!column) {
1583             break;
1584         }
1585
1586         datum = ovsdb_idl_read(row, column);
1587         if (column->type.key.type == OVSDB_TYPE_UUID &&
1588             column->type.key.u.uuid.refTableName) {
1589             struct cmd_show_table *ref_show;
1590             size_t j;
1591
1592             ref_show = cmd_show_find_table_by_name(
1593                 column->type.key.u.uuid.refTableName);
1594             if (ref_show) {
1595                 for (j = 0; j < datum->n; j++) {
1596                     const struct ovsdb_idl_row *ref_row;
1597
1598                     ref_row = ovsdb_idl_get_row_for_uuid(ctx->idl,
1599                                                          ref_show->table,
1600                                                          &datum->keys[j].uuid);
1601                     if (ref_row) {
1602                         cmd_show_row(ctx, ref_row, level + 1);
1603                     }
1604                 }
1605                 continue;
1606             }
1607         } else if (ovsdb_type_is_map(&column->type) &&
1608                    column->type.value.type == OVSDB_TYPE_UUID &&
1609                    column->type.value.u.uuid.refTableName) {
1610             struct cmd_show_table *ref_show;
1611             size_t j;
1612
1613             /* Prints the key to ref'ed table name map if the ref'ed table
1614              * is also defined in 'cmd_show_tables'.  */
1615             ref_show = cmd_show_find_table_by_name(
1616                 column->type.value.u.uuid.refTableName);
1617             if (ref_show && ref_show->name_column) {
1618                 ds_put_char_multiple(&ctx->output, ' ', (level + 1) * 4);
1619                 ds_put_format(&ctx->output, "%s:\n", column->name);
1620                 for (j = 0; j < datum->n; j++) {
1621                     const struct ovsdb_idl_row *ref_row;
1622
1623                     ref_row = ovsdb_idl_get_row_for_uuid(ctx->idl,
1624                                                          ref_show->table,
1625                                                          &datum->values[j].uuid);
1626
1627                     ds_put_char_multiple(&ctx->output, ' ', (level + 2) * 4);
1628                     ovsdb_atom_to_string(&datum->keys[j], column->type.key.type,
1629                                          &ctx->output);
1630                     ds_put_char(&ctx->output, '=');
1631                     if (ref_row) {
1632                         const struct ovsdb_datum *ref_datum;
1633
1634                         ref_datum = ovsdb_idl_read(ref_row,
1635                                                    ref_show->name_column);
1636                         ovsdb_datum_to_string(ref_datum,
1637                                               &ref_show->name_column->type,
1638                                               &ctx->output);
1639                     } else {
1640                         ds_put_cstr(&ctx->output, "\"<null>\"");
1641                     }
1642                     ds_put_char(&ctx->output, '\n');
1643                 }
1644                 continue;
1645             }
1646         }
1647
1648         if (!ovsdb_datum_is_default(datum, &column->type)) {
1649             ds_put_char_multiple(&ctx->output, ' ', (level + 1) * 4);
1650             ds_put_format(&ctx->output, "%s: ", column->name);
1651             ovsdb_datum_to_string(datum, &column->type, &ctx->output);
1652             ds_put_char(&ctx->output, '\n');
1653         }
1654     }
1655     show->recurse = false;
1656 }
1657
1658 static void
1659 cmd_show(struct ctl_context *ctx)
1660 {
1661     const struct ovsdb_idl_row *row;
1662
1663     for (row = ovsdb_idl_first_row(ctx->idl, cmd_show_tables[0].table);
1664          row; row = ovsdb_idl_next_row(row)) {
1665         cmd_show_row(ctx, row, 0);
1666     }
1667 }
1668
1669 \f
1670 /* Given pointer to dynamic array 'options_p',  array's current size
1671  * 'allocated_options_p' and number of added options 'n_options_p',
1672  * adds all command options to the array.  Enlarges the array if
1673  * necessary. */
1674 void
1675 ctl_add_cmd_options(struct option **options_p, size_t *n_options_p,
1676                     size_t *allocated_options_p, int opt_val)
1677 {
1678     struct option *o;
1679     const struct shash_node *node;
1680     size_t n_existing_options = *n_options_p;
1681
1682     SHASH_FOR_EACH (node, ctl_get_all_commands()) {
1683         const struct ctl_command_syntax *p = node->data;
1684
1685         if (p->options[0]) {
1686             char *save_ptr = NULL;
1687             char *name;
1688             char *s;
1689
1690             s = xstrdup(p->options);
1691             for (name = strtok_r(s, ",", &save_ptr); name != NULL;
1692                  name = strtok_r(NULL, ",", &save_ptr)) {
1693                 char *equals;
1694                 int has_arg;
1695
1696                 ovs_assert(name[0] == '-' && name[1] == '-' && name[2]);
1697                 name += 2;
1698
1699                 equals = strchr(name, '=');
1700                 if (equals) {
1701                     has_arg = required_argument;
1702                     *equals = '\0';
1703                 } else {
1704                     has_arg = no_argument;
1705                 }
1706
1707                 o = find_option(name, *options_p, *n_options_p);
1708                 if (o) {
1709                     ovs_assert(o - *options_p >= n_existing_options);
1710                     ovs_assert(o->has_arg == has_arg);
1711                 } else {
1712                     o = add_option(options_p, n_options_p, allocated_options_p);
1713                     o->name = xstrdup(name);
1714                     o->has_arg = has_arg;
1715                     o->flag = NULL;
1716                     o->val = opt_val;
1717                 }
1718             }
1719
1720             free(s);
1721         }
1722     }
1723     o = add_option(options_p, n_options_p, allocated_options_p);
1724     memset(o, 0, sizeof *o);
1725 }
1726
1727 /* Parses command-line input for commands. */
1728 struct ctl_command *
1729 ctl_parse_commands(int argc, char *argv[], struct shash *local_options,
1730                    size_t *n_commandsp)
1731 {
1732     struct ctl_command *commands;
1733     size_t n_commands, allocated_commands;
1734     int i, start;
1735
1736     commands = NULL;
1737     n_commands = allocated_commands = 0;
1738
1739     for (start = i = 0; i <= argc; i++) {
1740         if (i == argc || !strcmp(argv[i], "--")) {
1741             if (i > start) {
1742                 if (n_commands >= allocated_commands) {
1743                     struct ctl_command *c;
1744
1745                     commands = x2nrealloc(commands, &allocated_commands,
1746                                           sizeof *commands);
1747                     for (c = commands; c < &commands[n_commands]; c++) {
1748                         shash_moved(&c->options);
1749                     }
1750                 }
1751                 parse_command(i - start, &argv[start], local_options,
1752                               &commands[n_commands++]);
1753             } else if (!shash_is_empty(local_options)) {
1754                 ctl_fatal("missing command name (use --help for help)");
1755             }
1756             start = i + 1;
1757         }
1758     }
1759     if (!n_commands) {
1760         ctl_fatal("missing command name (use --help for help)");
1761     }
1762     *n_commandsp = n_commands;
1763     return commands;
1764 }
1765
1766 /* Prints all registered commands. */
1767 void
1768 ctl_print_commands(void)
1769 {
1770     const struct shash_node *node;
1771
1772     SHASH_FOR_EACH (node, ctl_get_all_commands()) {
1773         const struct ctl_command_syntax *p = node->data;
1774         char *options = xstrdup(p->options);
1775         char *options_begin = options;
1776         char *item;
1777
1778         for (item = strsep(&options, ","); item != NULL;
1779              item = strsep(&options, ",")) {
1780             if (item[0] != '\0') {
1781                 printf("[%s] ", item);
1782             }
1783         }
1784         printf(",%s,", p->name);
1785         print_command_arguments(p);
1786         printf("\n");
1787
1788         free(options_begin);
1789     }
1790
1791     exit(EXIT_SUCCESS);
1792 }
1793
1794 /* Given array of options 'options', prints them. */
1795 void
1796 ctl_print_options(const struct option *options)
1797 {
1798     for (; options->name; options++) {
1799         const struct option *o = options;
1800
1801         printf("--%s%s\n", o->name, o->has_arg ? "=ARG" : "");
1802         if (o->flag == NULL && o->val > 0 && o->val <= UCHAR_MAX) {
1803             printf("-%c%s\n", o->val, o->has_arg ? " ARG" : "");
1804         }
1805     }
1806
1807     exit(EXIT_SUCCESS);
1808 }
1809
1810 /* Returns the default local database path. */
1811 char *
1812 ctl_default_db(void)
1813 {
1814     static char *def;
1815     if (!def) {
1816         def = xasprintf("unix:%s/db.sock", ovs_rundir());
1817     }
1818     return def;
1819 }
1820
1821 /* Returns true if it looks like this set of arguments might modify the
1822  * database, otherwise false.  (Not very smart, so it's prone to false
1823  * positives.) */
1824 bool
1825 ctl_might_write_to_db(char **argv)
1826 {
1827     for (; *argv; argv++) {
1828         const struct ctl_command_syntax *p = shash_find_data(&all_commands, *argv);
1829         if (p && p->mode == RW) {
1830             return true;
1831         }
1832     }
1833     return false;
1834 }
1835
1836 void
1837 ctl_fatal(const char *format, ...)
1838 {
1839     char *message;
1840     va_list args;
1841
1842     va_start(args, format);
1843     message = xvasprintf(format, args);
1844     va_end(args);
1845
1846     vlog_set_levels(&VLM_db_ctl_base, VLF_CONSOLE, VLL_OFF);
1847     VLOG_ERR("%s", message);
1848     ovs_error(0, "%s", message);
1849     ctl_exit(EXIT_FAILURE);
1850 }
1851
1852 /* Frees the current transaction and the underlying IDL and then calls
1853  * exit(status).
1854  *
1855  * Freeing the transaction and the IDL is not strictly necessary, but it makes
1856  * for a clean memory leak report from valgrind in the normal case.  That makes
1857  * it easier to notice real memory leaks. */
1858 void
1859 ctl_exit(int status)
1860 {
1861     if (the_idl_txn) {
1862         ovsdb_idl_txn_abort(the_idl_txn);
1863         ovsdb_idl_txn_destroy(the_idl_txn);
1864     }
1865     ovsdb_idl_destroy(the_idl);
1866     exit(status);
1867 }
1868
1869 /* Command for showing overview of database contents. */
1870 static const struct ctl_command_syntax db_ctl_show_command[] = {
1871     {"show", 0, 0, "", pre_cmd_show, cmd_show, NULL, "", RO},
1872     {NULL, 0, 0, NULL, NULL, NULL, NULL, NULL, RO},
1873 };
1874
1875 /* Comman database commands to be registered. */
1876 static const struct ctl_command_syntax db_ctl_commands[] = {
1877     {"comment", 0, INT_MAX, "[ARG]...", NULL, NULL, NULL, "", RO},
1878     {"get", 2, INT_MAX, "TABLE RECORD [COLUMN[:KEY]]...",pre_cmd_get, cmd_get,
1879      NULL, "--if-exists,--id=", RO},
1880     {"list", 1, INT_MAX, "TABLE [RECORD]...", pre_cmd_list, cmd_list, NULL,
1881      "--if-exists,--columns=", RO},
1882     {"find", 1, INT_MAX, "TABLE [COLUMN[:KEY]=VALUE]...", pre_cmd_find,
1883      cmd_find, NULL, "--columns=", RO},
1884     {"set", 3, INT_MAX, "TABLE RECORD COLUMN[:KEY]=VALUE...", pre_cmd_set,
1885      cmd_set, NULL, "--if-exists", RW},
1886     {"add", 4, INT_MAX, "TABLE RECORD COLUMN [KEY=]VALUE...", pre_cmd_add,
1887      cmd_add, NULL, "--if-exists", RW},
1888     {"remove", 4, INT_MAX, "TABLE RECORD COLUMN KEY|VALUE|KEY=VALUE...",
1889      pre_cmd_remove, cmd_remove, NULL, "--if-exists", RW},
1890     {"clear", 3, INT_MAX, "TABLE RECORD COLUMN...", pre_cmd_clear, cmd_clear,
1891      NULL, "--if-exists", RW},
1892     {"create", 2, INT_MAX, "TABLE COLUMN[:KEY]=VALUE...", pre_create,
1893      cmd_create, post_create, "--id=", RW},
1894     {"destroy", 1, INT_MAX, "TABLE [RECORD]...", pre_cmd_destroy, cmd_destroy,
1895      NULL, "--if-exists,--all", RW},
1896     {"wait-until", 2, INT_MAX, "TABLE RECORD [COLUMN[:KEY]=VALUE]...",
1897      pre_cmd_wait_until, cmd_wait_until, NULL, "", RO},
1898     {NULL, 0, 0, NULL, NULL, NULL, NULL, NULL, RO},
1899 };
1900
1901 /* Registers commands represented by 'struct ctl_command_syntax's to
1902  * 'all_commands'.  The last element of 'commands' must be an all-NULL
1903  * element. */
1904 void
1905 ctl_register_commands(const struct ctl_command_syntax *commands)
1906 {
1907     const struct ctl_command_syntax *p;
1908
1909     for (p = commands; p->name; p++) {
1910         shash_add_assert(&all_commands, p->name, p);
1911     }
1912 }
1913
1914 /* Registers the 'db_ctl_commands' to 'all_commands'. */
1915 void
1916 ctl_init(const struct ctl_table_class tables_[])
1917 {
1918     tables = tables_;
1919     ctl_register_commands(db_ctl_commands);
1920     ctl_register_commands(db_ctl_show_command);
1921 }
1922
1923 /* Returns 'all_commands'. */
1924 const struct shash *
1925 ctl_get_all_commands(void)
1926 {
1927     return &all_commands;
1928 }
1929
1930 /* Returns the text for the database commands usage.  */
1931 const char *
1932 ctl_get_db_cmd_usage(void)
1933 {
1934     return "Database commands:\n\
1935   list TBL [REC]              list RECord (or all records) in TBL\n\
1936   find TBL CONDITION...       list records satisfying CONDITION in TBL\n\
1937   get TBL REC COL[:KEY]       print values of COLumns in RECord in TBL\n\
1938   set TBL REC COL[:KEY]=VALUE set COLumn values in RECord in TBL\n\
1939   add TBL REC COL [KEY=]VALUE add (KEY=)VALUE to COLumn in RECord in TBL\n\
1940   remove TBL REC COL [KEY=]VALUE  remove (KEY=)VALUE from COLumn\n\
1941   clear TBL REC COL           clear values from COLumn in RECord in TBL\n\
1942   create TBL COL[:KEY]=VALUE  create and initialize new record\n\
1943   destroy TBL REC             delete RECord from TBL\n\
1944   wait-until TBL REC [COL[:KEY]=VALUE]  wait until condition is true\n\
1945 Potentially unsafe database commands require --force option.\n";
1946 }
1947
1948 /* Initializes 'ctx' from 'command'. */
1949 void
1950 ctl_context_init_command(struct ctl_context *ctx,
1951                          struct ctl_command *command)
1952 {
1953     ctx->argc = command->argc;
1954     ctx->argv = command->argv;
1955     ctx->options = command->options;
1956
1957     ds_swap(&ctx->output, &command->output);
1958     ctx->table = command->table;
1959     ctx->try_again = false;
1960 }
1961
1962 /* Initializes the entire 'ctx'. */
1963 void
1964 ctl_context_init(struct ctl_context *ctx, struct ctl_command *command,
1965                  struct ovsdb_idl *idl, struct ovsdb_idl_txn *txn,
1966                  struct ovsdb_symbol_table *symtab,
1967                  void (*invalidate_cache)(struct ctl_context *))
1968 {
1969     if (command) {
1970         ctl_context_init_command(ctx, command);
1971     }
1972     ctx->idl = idl;
1973     ctx->txn = txn;
1974     ctx->symtab = symtab;
1975     ctx->invalidate_cache = invalidate_cache;
1976 }
1977
1978 /* Completes processing of 'command' within 'ctx'. */
1979 void
1980 ctl_context_done_command(struct ctl_context *ctx,
1981                          struct ctl_command *command)
1982 {
1983     ds_swap(&ctx->output, &command->output);
1984     command->table = ctx->table;
1985 }
1986
1987 /* Finishes up with 'ctx'.
1988  *
1989  * If command is nonnull, first calls ctl_context_done_command() to complete
1990  * processing that command within 'ctx'. */
1991 void
1992 ctl_context_done(struct ctl_context *ctx,
1993                  struct ctl_command *command)
1994 {
1995     if (command) {
1996         ctl_context_done_command(ctx, command);
1997     }
1998     invalidate_cache(ctx);
1999 }
2000
2001 /* Finds and returns the "struct ctl_table_class *" with 'table_name' by
2002  * searching the 'tables'. */
2003 static const struct ctl_table_class *
2004 get_table(const char *table_name)
2005 {
2006     const struct ctl_table_class *table;
2007     const struct ctl_table_class *best_match = NULL;
2008     unsigned int best_score = 0;
2009
2010     for (table = tables; table->class; table++) {
2011         unsigned int score = score_partial_match(table->class->name,
2012                                                  table_name);
2013         if (score > best_score) {
2014             best_match = table;
2015             best_score = score;
2016         } else if (score == best_score) {
2017             best_match = NULL;
2018         }
2019     }
2020     if (best_match) {
2021         return best_match;
2022     } else if (best_score) {
2023         ctl_fatal("multiple table names match \"%s\"", table_name);
2024     } else {
2025         ctl_fatal("unknown table \"%s\"", table_name);
2026     }
2027     return NULL;
2028 }
2029
2030 /* Sets the column of 'row' in 'table'. */
2031 static void
2032 set_column(const struct ctl_table_class *table,
2033            const struct ovsdb_idl_row *row, const char *arg,
2034            struct ovsdb_symbol_table *symtab)
2035 {
2036     const struct ovsdb_idl_column *column;
2037     char *key_string, *value_string;
2038     char *error;
2039
2040     error = parse_column_key_value(arg, table, &column, &key_string,
2041                                    NULL, NULL, 0, &value_string);
2042     die_if_error(error);
2043     if (!value_string) {
2044         ctl_fatal("%s: missing value", arg);
2045     }
2046     check_mutable(row, column);
2047
2048     if (key_string) {
2049         union ovsdb_atom key, value;
2050         struct ovsdb_datum datum;
2051
2052         if (column->type.value.type == OVSDB_TYPE_VOID) {
2053             ctl_fatal("cannot specify key to set for non-map column %s",
2054                       column->name);
2055         }
2056
2057         die_if_error(ovsdb_atom_from_string(&key, &column->type.key,
2058                                             key_string, symtab));
2059         die_if_error(ovsdb_atom_from_string(&value, &column->type.value,
2060                                             value_string, symtab));
2061
2062         ovsdb_datum_init_empty(&datum);
2063         ovsdb_datum_add_unsafe(&datum, &key, &value, &column->type);
2064
2065         ovsdb_atom_destroy(&key, column->type.key.type);
2066         ovsdb_atom_destroy(&value, column->type.value.type);
2067
2068         ovsdb_datum_union(&datum, ovsdb_idl_read(row, column),
2069                           &column->type, false);
2070         ovsdb_idl_txn_verify(row, column);
2071         ovsdb_idl_txn_write(row, column, &datum);
2072     } else {
2073         struct ovsdb_datum datum;
2074
2075         die_if_error(ovsdb_datum_from_string(&datum, &column->type,
2076                                              value_string, symtab));
2077         ovsdb_idl_txn_write(row, column, &datum);
2078     }
2079
2080     free(key_string);
2081     free(value_string);
2082 }
2083
2084 void ctl_set_column(const char *table_name,
2085                     const struct ovsdb_idl_row *row, const char *arg,
2086                     struct ovsdb_symbol_table *symtab)
2087 {
2088     set_column(get_table(table_name), row, arg, symtab);
2089 }