Ticket #48755 - moving an entry could make the online init fail

Bug Description: Online init (aka Total update, bulk import) scans the
primary id2entry db in the order of ID.  If Entry A is moved under a
new superior Entry B which was generated after Entry A, when Entry A
is sent to a consumer using online init, its parent entry does not
exist on the consumer and the online init fails.

Fix Description:
- Added a command BACK_INFO_IS_ENTRYRDN to slapi_back_get_info, which
  returns the status of entryrdn switch maintained in the backend.

- If slapi_backend_get_info(BACK_INFO_IS_ENTRYRDN) returns true for
  the replicated backend, repl5_tot_run searches the entry with the
  filter:
    (parentid>=1)
  instead of:
    (|(objectclass=ldapsubentry)(objectclass=nstombstone)(nsuniqueid=*))".
  The filter MUST be this single range filter since the IDList created
  for this bulk-import is not sorted by ID, but in the order of parentid.

- Added a new flag OP_FLAG_BULK_IMPORT to the operation.  Setting the
  flag implies to skip the internal operations such as eliminating
  ldapsubentry from the search results or a filter check.

- In addition, idl_new_range_fetch is modified so that ...
  * A range search for parentid ignores nsslapd-idlistscanlimit by
    setting SLAPI_OP_RANGE_NO_ALLIDS as well as it skips sorting the
	IDlist by ID by setting SLAPI_OP_RANGE_NO_IDL_SORT.
  * In case SLAPI_OP_RANGE_NO_IDL_SORT is set, idl_new_range_fetch
    checks whether the key (in this case parentid) is in the IDlist.
	If it exists, the ID is appended.  If it does not, the ID is in
	the leftover list and appended when the parent ID is found in the
	IDlist.

- Increased the version of rdn-format-# in DBVERSION to 3.

- Upgrade script 91reindex.pl.in is added which reindex the parentid
  index file in the integer order if the version of rdn-format-# in
  DBVERSION is less than 3.

- Notes: In case entryrdn is used (nsslapd-subtree-rename-switch: on),
  all the type of entries including the tombstone entry and the ldap-
  subentry are indexed with parentid.

- Another notes: This special treatment is enabled only when nsslapd-
  subtree-rename-switch is on.  If entrydn is used instead of entryrdn,
  there is no possibility to establish the order that child comes first.
  Plus, to replace nsslapd-subtree-rename-switch on with off, the main
  db file id2entry.db needs to be recreated so that each entry stores
  DN in it.  It requires export then import, which corrects the order
  of the entries and the order problem disappears.

https://fedorahosted.org/389/ticket/48755

Reviewed by mreynolds@redhat and [email protected] (Thank you, Mark and William!!)
Also, thanks to [email protected] and [email protected] for their discussion!

Makefile.am

@@ -841,7 +841,8 @@ update_DATA = ldap/admin/src/scripts/exampleupdate.pl \
 	ldap/admin/src/scripts/50AES-pbe-plugin.ldif\
 	ldap/admin/src/scripts/50updateconfig.ldif \
 	ldap/admin/src/scripts/52updateAESplugin.pl \
-	ldap/admin/src/scripts/dnaplugindepends.ldif
+	ldap/admin/src/scripts/dnaplugindepends.ldif \
+	ldap/admin/src/scripts/91reindex.pl
 
 update_SCRIPTS = ldap/admin/src/scripts/exampleupdate.sh
 

Makefile.in

@@ -2259,7 +2259,8 @@ update_DATA = ldap/admin/src/scripts/exampleupdate.pl \
 	ldap/admin/src/scripts/50AES-pbe-plugin.ldif\
 	ldap/admin/src/scripts/50updateconfig.ldif \
 	ldap/admin/src/scripts/52updateAESplugin.pl \
-	ldap/admin/src/scripts/dnaplugindepends.ldif
+	ldap/admin/src/scripts/dnaplugindepends.ldif \
+	ldap/admin/src/scripts/91reindex.pl
 
 update_SCRIPTS = ldap/admin/src/scripts/exampleupdate.sh
 

ldap/admin/src/scripts/91reindex.pl.in

@@ -0,0 +1,103 @@
+use Mozilla::LDAP::Conn;
+use Mozilla::LDAP::Utils qw(normalizeDN);
+use Mozilla::LDAP::API qw(:constant ldap_url_parse ldap_explode_dn);
+use DSUpdate qw(isOffline);
+
+sub runinst {
+    my ($inf, $inst, $dseldif, $conn) = @_;
+    my $rc, @errs;
+
+    # List of index to be reindexed
+    my @toreindex = qw(parentid);
+    # rdn-format value.  See $rdn_format set below.
+    # If equal to or greater than this value, no need to reindex.
+    # If it needs to be unconditionally reindexed, set 0.
+    my @rdnconditions = (4)
+
+    my $config = $conn->search("cn=config", "base", "(objectclass=*)");
+    if (!$config) {
+        push @errs, ['error_finding_config_entry', 'cn=config',
+                     $conn->getErrorString()];
+        return @errs;
+    }
+
+    ($rc, @errs) = isOffline($inf, $inst, $conn);
+    if (!$rc) {
+        return @errs;
+    }
+
+    my $reindex = "@sbindir@/db2index -Z $inst";
+    my @errs;
+    my $instconf = $conn->search("cn=ldbm database,cn=plugins,cn=config", "onelevel", "(objectclass=*)");
+    if (!$instconf) {
+        push @errs, ['error_finding_config_entry', 'cn=*,cn=ldbm database,cn=plugins,cn=config', $conn->getErrorString()];
+        return @errs;
+    }
+
+    my $dbconf = $conn->search("cn=config,cn=ldbm database,cn=plugins,cn=config", "base", "(objectclass=*)");
+    if (!$dbconf) {
+        push @errs, ['error_finding_config_entry',
+                     'cn=config,cn=ldbm database,cn=plugins,cn=config',
+                     $conn->getErrorString()];
+        return @errs;
+    }
+
+    # Get the value of nsslapd-subtree-rename-switch.
+    my $switch = $dbconf->getValues('nsslapd-subtree-rename-switch');
+    if ("" eq $switch) {
+        return (); # subtree-rename-switch does not exist; do nothing.
+    } elsif ("off" eq $switch || "OFF" eq $switch) {
+        return (); # subtree-rename-switch is OFF; do nothing.
+    }
+
+    my $dbdir = $dbconf->getValues('nsslapd-directory');
+    my $dbversion0 = $dbdir . "/DBVERSION";
+    my $rdn_format = 0;
+    my $dbversionstr = "";
+    if (!open(DBVERSION, "$dbversion0")) {
+        push @errs, ['error_opening_file', $dbversion0, $!];
+        return @errs;
+    } else {
+        while (<DBVERSION>) {
+            if ($_ =~ /rdn-format/) {
+                $rdn_format = 1;
+                $dbversionstr = $_;
+                if ($_ =~ /rdn-format-1/) {
+                    $rdn_format = 2;
+                } elsif ($_ =~ /rdn-format-2/) {
+                    $rdn_format = 3;
+                } elsif ($_ =~ /rdn-format-3/) {
+                    $rdn_format = 4;
+                } elsif ($_ =~ /rdn-format-4/) {
+                    $rdn_format = 5;
+                } elsif ($_ =~ /rdn-format-5/) {
+                    $rdn_format = 6;
+                } elsif ($_ =~ /rdn-format-/) {
+                    # assume greater than -5
+                    $rdn_format = 7;
+                }
+            }
+        }
+        close DBVERSION;
+    }
+
+    while ($instconf) {
+        my $backend= $instconf->getValues('cn');
+        if (($backend eq "config") || ($backend eq "monitor")) {
+            goto NEXT;
+        }
+
+        for (my $idx = 0; $ <= $#toreindex; $idx++) {
+            if (0 == $rdnconditions[$idx] || $rdnconditions[$idx] > $rdn_format) {
+                my $rc = system("$reindex -n $backend -t $idx");
+                if ($rc) {
+                    push @errs, ["error_reindexng", $idx, $backend, $rc];
+                }
+            }
+        }
+NEXT:
+        $instconf = $conn->nextEntry();
+    }
+
+    return @errs;
+}

ldap/admin/src/scripts/91subtreereindex.pl

@@ -51,14 +51,18 @@ sub runinst {
                 if ($_ =~ /rdn-format-1/) {
                     $is_rdn_format = 2;
                 }
-                if ($_ =~ /rdn-format-2/) {
+                elsif ($_ =~ /rdn-format-2/) {
                     $is_rdn_format = 3;
                 }
+                elsif ($_ =~ /rdn-format-/) {
+                    # assume greater than -2
+                    $is_rdn_format = 4;
+                }
             }
         }
         close DBVERSION;
 
-        if (3 == $is_rdn_format) {
+        if (3 <= $is_rdn_format) {
             return (); # DB already has the new rdn format.
         }
 

ldap/admin/src/scripts/setup-ds.res.in

@@ -209,3 +209,4 @@ error_opening_file = Opening file '%s' failed.  Error: %s\n
 error_format_error = '%s' has invalid format.\n
 error_update_not_offline = Error: offline mode selected but the server [%s] is still running.\n
 error_update_all = Failed to update all the Directory Server instances.\n
+error_reindexing = Failed to reindex '%s' in backend '%s'.  Error: %s\n

ldap/ldif/template-dse.ldif.in

@@ -930,6 +930,7 @@ objectclass: nsIndex
 cn: parentid
 nssystemindex: true
 nsindextype: eq
+nsmatchingrule: integerOrderingMatch
 
 dn: cn=seeAlso,cn=default indexes, cn=config,cn=ldbm database,cn=plugins,cn=config
 objectclass: top

ldap/servers/plugins/replication/repl5_replica_config.c

@@ -405,12 +405,12 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                 {
                     if (apply_mods)
                         replica_set_precise_purging(r, 0);
-                    }
+                }
                 else if (strcasecmp (config_attr, type_replicaReleaseTimeout) == 0 )
                 {
                     if (apply_mods)
                         replica_set_release_timeout(r, 0);
-                    }
+                }
                 else
                 {
                     *returncode = LDAP_UNWILLING_TO_PERFORM;
@@ -572,8 +572,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                 {
                     if (apply_mods)
                     {
-                        if (apply_mods && config_attr_value[0])
-                        {
+                        if (config_attr_value[0]) {
                             PRUint64 on_off = 0;
 
                             if (strcasecmp(config_attr_value, "on") == 0){
@@ -592,7 +591,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                                 break;
                             }
                             replica_set_precise_purging(r, on_off);
-                        } else if (apply_mods) {
+                        } else {
                             replica_set_precise_purging(r, 0);
                         }
                     }

ldap/servers/plugins/replication/repl5_tot_protocol.c

@@ -323,6 +323,10 @@ repl5_tot_run(Private_Repl_Protocol *prp)
 	int init_retry = 0;
 	Replica *replica;
 	ReplicaId rid = 0; /* Used to create the replica keep alive subentry */
+	Slapi_Entry *suffix = NULL;
+	char **instances = NULL;
+	Slapi_Backend *be = NULL;
+	int is_entryrdn = 0;
 	
 	PR_ASSERT(NULL != prp);
 
@@ -354,21 +358,21 @@ retry:
      */
     if (rc != ACQUIRE_SUCCESS)
     {
-		int optype, ldaprc, wait_retry;
-		conn_get_error(prp->conn, &optype, &ldaprc);
-		if (rc == ACQUIRE_TRANSIENT_ERROR && INIT_RETRY_MAX > init_retry++) {
-			wait_retry = init_retry * INIT_RETRY_INT;
-			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
-						"acquire replica for total update, error: %d,"
+        int optype, ldaprc, wait_retry;
+        conn_get_error(prp->conn, &optype, &ldaprc);
+        if (rc == ACQUIRE_TRANSIENT_ERROR && INIT_RETRY_MAX > init_retry++) {
+            wait_retry = init_retry * INIT_RETRY_INT;
+            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
+                        "acquire replica for total update, error: %d,"
                                                 " retrying in %d seconds.\n",
-						 ldaprc, wait_retry);
-			DS_Sleep(PR_SecondsToInterval(wait_retry));
-			goto retry;
-		} else {
-			agmt_set_last_init_status(prp->agmt, ldaprc,
-				  prp->last_acquire_response_code, 0, NULL);
-			goto done;
-                }
+                         ldaprc, wait_retry);
+            DS_Sleep(PR_SecondsToInterval(wait_retry));
+            goto retry;
+        } else {
+            agmt_set_last_init_status(prp->agmt, ldaprc,
+                  prp->last_acquire_response_code, 0, NULL);
+            goto done;
+        }
     }
 	else if (prp->terminate)
     {
@@ -405,48 +409,121 @@ retry:
        and that the order implies that perent entry is always ahead of the
        child entry in the list. Otherwise, the consumer would not be
        properly updated because bulk import at the moment skips orphand entries. */
-	/* XXXggood above assumption may not be valid if orphaned entry moved???? */
+    /* XXXggood above assumption may not be valid if orphaned entry moved???? */
 
     agmt_set_last_init_status(prp->agmt, 0, 0, 0, "Total update in progress");
 
     slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Beginning total update of replica "
-		"\"%s\".\n", agmt_get_long_name(prp->agmt));
+                    "\"%s\".\n", agmt_get_long_name(prp->agmt));
 
     /* RMREPL - need to send schema here */
 
     pb = slapi_pblock_new ();
 
-    /* we need to provide managedsait control so that referral entries can
-       be replicated */
-    ctrls = (LDAPControl **)slapi_ch_calloc (3, sizeof (LDAPControl *));
-    ctrls[0] = create_managedsait_control ();
-    ctrls[1] = create_backend_control(area_sdn);
+    replica = (Replica*) object_get_data(prp->replica_object);
+    /*
+     * Get the info about the entryrdn vs. entrydn from the backend.
+     * If NOT is_entryrdn, its ancestor entries are always found prior to an entry.
+     */
+    rc = slapi_lookup_instance_name_by_suffix((char *)slapi_sdn_get_dn(area_sdn), NULL, &instances, 1);
+    if (rc || !instances) {
+        slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
+                        "get the instance name for the suffix \"%s\".\n", slapi_sdn_get_dn(area_sdn));
+        goto done;
+    }
+    be = slapi_be_select_by_instance_name(instances[0]);
+    if (!be) {
+        slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
+                        "get the instance for the suffix \"%s\".\n", slapi_sdn_get_dn(area_sdn));
+        goto done;
+    }
+    rc = slapi_back_get_info(be, BACK_INFO_IS_ENTRYRDN, (void **)&is_entryrdn);
+    if (is_entryrdn) {
+        /*
+         * Supporting entries out of order -- parent could have a larger id than its children.
+         * Entires are retireved sorted by parentid without the allid threshold.
+         */
+        /* Get suffix */
+        rc = slapi_search_internal_get_entry(area_sdn, NULL, &suffix, repl_get_plugin_identity(PLUGIN_MULTIMASTER_REPLICATION));
+        if (rc) {
+            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
+                            "get the suffix entry \"%s\".\n", slapi_sdn_get_dn(area_sdn));
+            goto done;
+        }
 
-	/* Time to make sure it exists a keep alive subentry for that replica */
-	replica = (Replica*) object_get_data(prp->replica_object);
-	if (replica)
-	{
-		rid = replica_get_rid(replica);
-	}
-	replica_subentry_check(area_sdn, rid);
-
-    slapi_search_internal_set_pb (pb, slapi_sdn_get_dn (area_sdn), 
-                                  LDAP_SCOPE_SUBTREE, "(|(objectclass=ldapsubentry)(objectclass=nstombstone)(nsuniqueid=*))", NULL, 0, ctrls, NULL, 
-                                  repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), 0);
-
-	cb_data.prp = prp;
-	cb_data.rc = 0;
-	cb_data.num_entries = 0UL;
-	cb_data.sleep_on_busy = 0UL;
-	cb_data.last_busy = current_time ();
-	cb_data.flowcontrol_detection = 0;
-	cb_data.lock = PR_NewLock();
-
-	/* This allows during perform_operation to check the callback data
-	 * especially to do flow contol on delta send msgid / recv msgid
-	 */
-	conn_set_tot_update_cb(prp->conn, (void *) &cb_data);
+        cb_data.prp = prp;
+        cb_data.rc = 0;
+        cb_data.num_entries = 1UL;
+        cb_data.sleep_on_busy = 0UL;
+        cb_data.last_busy = current_time ();
+        cb_data.flowcontrol_detection = 0;
+        cb_data.lock = PR_NewLock();
+
+        /* This allows during perform_operation to check the callback data
+         * especially to do flow contol on delta send msgid / recv msgid
+         */
+        conn_set_tot_update_cb(prp->conn, (void *) &cb_data);
+
+        /* Send suffix first. */
+        rc = send_entry(suffix, (void *)&cb_data);
+        if (rc) {
+            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Warning: unable to "
+                            "send the suffix entry \"%s\" to the consumer.\n", slapi_sdn_get_dn(area_sdn));
+            goto done;
+        }
+
+        /* we need to provide managedsait control so that referral entries can
+           be replicated */
+        ctrls = (LDAPControl **)slapi_ch_calloc (3, sizeof (LDAPControl *));
+        ctrls[0] = create_managedsait_control ();
+        ctrls[1] = create_backend_control(area_sdn);
+
+        /* Time to make sure it exists a keep alive subentry for that replica */
+        if (replica)
+        {
+            rid = replica_get_rid(replica);
+        }
+        replica_subentry_check(area_sdn, rid);
     
+        /* Send the subtree of the suffix in the order of parentid index plus ldapsubentry and nstombstone. */
+        slapi_search_internal_set_pb(pb, slapi_sdn_get_dn (area_sdn),
+                                     LDAP_SCOPE_SUBTREE, "(parentid>=1)", NULL, 0, ctrls, NULL,
+                                     repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), OP_FLAG_BULK_IMPORT);
+        cb_data.num_entries = 0UL;
+    } else {
+        /* Original total update */
+        /* we need to provide managedsait control so that referral entries can
+           be replicated */
+        ctrls = (LDAPControl **)slapi_ch_calloc (3, sizeof (LDAPControl *));
+        ctrls[0] = create_managedsait_control ();
+        ctrls[1] = create_backend_control(area_sdn);
+
+        /* Time to make sure it exists a keep alive subentry for that replica */
+        replica = (Replica*) object_get_data(prp->replica_object);
+        if (replica)
+        {
+            rid = replica_get_rid(replica);
+        }
+        replica_subentry_check(area_sdn, rid);
+
+        slapi_search_internal_set_pb (pb, slapi_sdn_get_dn (area_sdn),
+                                      LDAP_SCOPE_SUBTREE, "(|(objectclass=ldapsubentry)(objectclass=nstombstone)(nsuniqueid=*))", NULL, 0, ctrls, NULL,
+                                      repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), 0);
+
+        cb_data.prp = prp;
+        cb_data.rc = 0;
+        cb_data.num_entries = 0UL;
+        cb_data.sleep_on_busy = 0UL;
+        cb_data.last_busy = current_time ();
+        cb_data.flowcontrol_detection = 0;
+        cb_data.lock = PR_NewLock();
+
+        /* This allows during perform_operation to check the callback data
+         * especially to do flow contol on delta send msgid / recv msgid
+         */
+        conn_set_tot_update_cb(prp->conn, (void *) &cb_data);
+    }
+
 	/* Before we get started on sending entries to the replica, we need to 
 	 * setup things for async propagation: 
 	 * 1. Create a thread that will read the LDAP results from the connection.
@@ -470,7 +547,7 @@ retry:
     slapi_search_internal_callback_pb (pb, &cb_data /* callback data */,
                                        get_result /* result callback */,
                                        send_entry /* entry callback */,
-	    					           NULL /* referral callback*/);
+                                       NULL /* referral callback*/);
 
 	/* 
 	 * After completing the sending operation (or optionally failing), we need to clean up

ldap/servers/slapd/back-ldbm/back-ldbm.h

@@ -132,7 +132,7 @@ typedef unsigned short u_int16_t;
 #define BDB_BACKEND     "libback-ldbm" /* This backend plugin */
 #define BDB_NEWIDL      "newidl"       /* new idl format */
 #define BDB_RDNFORMAT   "rdn-format"   /* Subtree rename enabled */
-#define BDB_RDNFORMAT_VERSION   "2"    /* rdn-format version (by default, 0) */
+#define BDB_RDNFORMAT_VERSION   "3"    /* rdn-format version (by default, 0) */
 #define BDB_DNFORMAT    "dn-4514"      /* DN format RFC 4514 compliant */
 #define BDB_DNFORMAT_VERSION    "1"    /* DN format version */
 
@@ -808,11 +808,11 @@ typedef struct _back_search_result_set
 /* #define LDBM_ENTRYRDN_OID			"2.16.840.1.113730.3.1.2097" */
 
 #define LDBM_ANCESTORID_STR         "ancestorid"
-#define LDBM_ENTRYDN_STR            "entrydn"
+#define LDBM_ENTRYDN_STR            SLAPI_ATTR_ENTRYDN
 #define LDBM_ENTRYRDN_STR           "entryrdn"
 #define LDBM_NUMSUBORDINATES_STR    "numsubordinates"
 #define LDBM_TOMBSTONE_NUMSUBORDINATES_STR    "tombstonenumsubordinates"
-#define LDBM_PARENTID_STR           "parentid"
+#define LDBM_PARENTID_STR           SLAPI_ATTR_PARENTID
 
 /* Name of psuedo attribute used to track default indexes */
 #define LDBM_PSEUDO_ATTR_DEFAULT	".default"

ldap/servers/slapd/back-ldbm/dblayer.c

@@ -7315,6 +7315,11 @@ ldbm_back_get_info(Slapi_Backend *be, int cmd, void **info)
         }
         break;
     }
+    case BACK_INFO_IS_ENTRYRDN:
+    {
+        *(int *)info = entryrdn_get_switch();
+        break;
+    }
     default:
         break;
     }

ldap/servers/slapd/back-ldbm/filterindex.c

@@ -552,6 +552,8 @@ range_candidates(
     struct berval *low = NULL, *high = NULL;
     struct berval **lows = NULL, **highs = NULL;
     back_txn txn = {NULL};
+    int operator = 0;
+    Operation *op = NULL;
 
     LDAPDebug(LDAP_DEBUG_TRACE, "=> range_candidates attr=%s\n", type, 0, 0);
 
@@ -579,17 +581,24 @@ range_candidates(
         high = attr_value_lowest(highs, slapi_berval_cmp);
     }
 
+    /* Check if it is for bulk import. */
+    slapi_pblock_get(pb, SLAPI_OPERATION, &op);
+    if (entryrdn_get_switch() && operation_is_flag_set(op, OP_FLAG_INTERNAL) &&
+        operation_is_flag_set(op, OP_FLAG_BULK_IMPORT)) {
+        /* parentid is treated specially that is needed for the bulk import. (See #48755) */
+        operator = SLAPI_OP_RANGE_NO_IDL_SORT|SLAPI_OP_RANGE_NO_ALLIDS;
+    }
     if (low == NULL) {
-        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY,
-                                   SLAPI_OP_LESS_OR_EQUAL,
+        operator |= SLAPI_OP_LESS_OR_EQUAL;
+        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY, operator,
                                    high, NULL, 0, &txn, err, allidslimit);
     } else if (high == NULL) {
-        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY,
-                                   SLAPI_OP_GREATER_OR_EQUAL,
+        operator |= SLAPI_OP_GREATER_OR_EQUAL;
+        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY, operator,
                                    low, NULL, 0, &txn, err, allidslimit);
     } else {
-        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY,
-                                   SLAPI_OP_LESS_OR_EQUAL,
+        operator |= SLAPI_OP_LESS_OR_EQUAL;
+        idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY, operator,
                                    low, high, 1, &txn, err, allidslimit);
     }
 

ldap/servers/slapd/back-ldbm/idl_new.c

@@ -350,17 +350,31 @@ error:
     return idl;
 }
 
+typedef struct _range_id_pair {
+    ID key;
+    ID id;
+} idl_range_id_pair;
 /* 
  * Perform the range search in the idl layer instead of the index layer
  * to improve the performance.
  */
+/*
+ * NOTE:
+ * In the total update (bulk import), an entry requires its ancestors already added.
+ * To guarantee it, the range search with parentid is used with setting the flag
+ * SLAPI_OP_RANGE_NO_IDL_SORT in operator.
+ *
+ * If the flag is set,
+ * 1. the IDList is not sorted by the ID.
+ * 2. holding to add an ID to the IDList unless the key is found in the IDList.
+ */
 IDList *
 idl_new_range_fetch(
-    backend *be, 
-    DB* db, 
-    DBT *lowerkey, 
+    backend *be,
+    DB* db,
+    DBT *lowerkey,
     DBT *upperkey,
-    DB_TXN *txn, 
+    DB_TXN *txn,
     struct attrinfo *ai,
     int *flag_err,
     int allidslimit,
@@ -380,7 +394,7 @@ idl_new_range_fetch(
     size_t count = 0;
 #ifdef DB_USE_BULK_FETCH
     /* beware that a large buffer on the stack might cause a stack overflow on some platforms */
-    char buffer[BULK_FETCH_BUFFER_SIZE]; 
+    char buffer[BULK_FETCH_BUFFER_SIZE];
     void *ptr;
     DBT dataret;
 #endif
@@ -388,15 +402,21 @@ idl_new_range_fetch(
     struct ldbminfo *li = (struct ldbminfo *)be->be_database->plg_private;
     time_t curtime;
     void *saved_key = NULL;
+    int coreop = operator & SLAPI_OP_RANGE;
+    ID key;
+    ID suffix;
+    idl_range_id_pair *leftover = NULL;
+    size_t leftoverlen = 32;
+    int leftovercnt = 0;
 
     if (NULL == flag_err) {
         return NULL;
     }
 
-    *flag_err = 0;
     if (NEW_IDL_NOOP == *flag_err) {
         return NULL;
     }
+
     dblayer_txn_init(li, &s_txn);
     if (txn) {
         dblayer_read_txn_begin(be, txn, &s_txn);
@@ -460,7 +480,7 @@ idl_new_range_fetch(
 #ifdef DB_USE_BULK_FETCH
     while (cur_key.data &&
            (upperkey && upperkey->data ?
-            ((operator == SLAPI_OP_LESS) ?
+            ((coreop == SLAPI_OP_LESS) ?
              DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) < 0 :
              DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) <= 0) :
             PR_TRUE /* e.g., (x > a) */)) {
@@ -496,6 +516,9 @@ idl_new_range_fetch(
                 goto error;
             }
         }
+        if (operator & SLAPI_OP_RANGE_NO_IDL_SORT) {
+            key = (ID)strtol((char *)cur_key.data+1 , (char **)NULL, 10);
+        }
         while (PR_TRUE) {
             DB_MULTIPLE_NEXT(ptr, &data, dataret.data, dataret.size);
             if (dataret.data == NULL) break;
@@ -524,7 +547,29 @@ idl_new_range_fetch(
             /* note the last id read to check for dups */
             lastid = id;
             /* we got another ID, add it to our IDL */
-            idl_rc = idl_append_extend(&idl, id);
+            if (operator & SLAPI_OP_RANGE_NO_IDL_SORT) {
+                if (!idl) {
+                    /* First time.  Keep the suffix ID. */
+                    suffix = key;
+                    idl_rc = idl_append_extend(&idl, id);
+                } else if ((key == suffix) || idl_id_is_in_idlist(idl, key)) {
+                    /* the parent is the suffix or already in idl. */
+                    idl_rc = idl_append_extend(&idl, id);
+                } else {
+                    /* Otherwise, keep the {key,id} in leftover array */
+                    if (!leftover) {
+                        leftover = (idl_range_id_pair *)slapi_ch_calloc(leftoverlen, sizeof(idl_range_id_pair));
+                    } else if (leftovercnt == leftoverlen) {
+                        leftover = (idl_range_id_pair *)slapi_ch_realloc((char *)leftover, 2 * leftoverlen * sizeof(idl_range_id_pair));
+                        memset(leftover + leftovercnt, 0, leftoverlen);
+                        leftoverlen *= 2;
+                    }
+                    leftover[leftovercnt].key = key;
+                    leftover[leftovercnt++].id = id;
+                }
+            } else {
+                idl_rc = idl_append_extend(&idl, id);
+            }
             if (idl_rc) {
                 LDAPDebug1Arg(LDAP_DEBUG_ANY,
                               "unable to extend id list (err=%d)\n", idl_rc);
@@ -581,7 +626,7 @@ idl_new_range_fetch(
     }
 #else
     while (upperkey && upperkey->data ?
-           ((operator == SLAPI_OP_LESS) ?
+           ((coreop == SLAPI_OP_LESS) ?
             DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) < 0 :
             DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) <= 0) :
            PR_TRUE /* e.g., (x > a) */) {
@@ -698,9 +743,27 @@ error:
     *flag_err = ret;
 
     /* sort idl */
-    if (idl && !ALLIDS(idl)) {
-        qsort((void *)&idl->b_ids[0], idl->b_nids,
-              (size_t)sizeof(ID), idl_sort_cmp);
+    if (idl && !ALLIDS(idl) && !(operator & SLAPI_OP_RANGE_NO_IDL_SORT)) {
+        qsort((void *)&idl->b_ids[0], idl->b_nids, (size_t)sizeof(ID), idl_sort_cmp);
+    }
+    if (operator & SLAPI_OP_RANGE_NO_IDL_SORT) {
+        int i;
+        int left = leftovercnt;
+        while (left) {
+            for (i = 0; i < leftovercnt; i++) {
+                if (leftover[i].key && idl_id_is_in_idlist(idl, leftover[i].key)) {
+                    idl_rc = idl_append_extend(&idl, leftover[i].id);
+                    if (idl_rc) {
+                        LDAPDebug1Arg(LDAP_DEBUG_ANY, "unable to extend id list (err=%d)\n", idl_rc);
+                        idl_free(&idl);
+                        return NULL;
+                    }
+                    leftover[i].key = 0;
+                    left--;
+                }
+            }
+        }
+        slapi_ch_free((void **)&leftover);
     }
     return idl;
 }

ldap/servers/slapd/back-ldbm/index.c

@@ -1232,6 +1232,7 @@ index_range_read_ext(
     int timelimit = -1;
     back_search_result_set *sr = NULL;
     int isroot = 0;
+    int coreop = operator & SLAPI_OP_RANGE;
 
     if (!pb) {
         LDAPDebug(LDAP_DEBUG_ANY, "index_range_read: NULL pblock\n",
@@ -1278,7 +1279,7 @@ index_range_read_ext(
     LDAPDebug1Arg(LDAP_DEBUG_TRACE, "index_range_read lookthrough_limit=%d\n",
                   lookthrough_limit);
 
-    switch( operator ) {
+    switch( coreop ) {
       case SLAPI_OP_LESS:
       case SLAPI_OP_LESS_OR_EQUAL:
       case SLAPI_OP_GREATER_OR_EQUAL:
@@ -1287,7 +1288,7 @@ index_range_read_ext(
       default:
         LDAPDebug( LDAP_DEBUG_ANY,
               "<= index_range_read(%s,%s) NULL (operator %i)\n",
-              type, prefix, operator );
+              type, prefix, coreop );
         index_free_prefix(prefix);
         return( NULL );
     }
@@ -1343,7 +1344,7 @@ index_range_read_ext(
     if (range != 1) { /* open range search */
         char *tmpbuf = NULL;
         /* this is a search with only one boundary value */
-        switch( operator ) {
+        switch( coreop ) {
           case SLAPI_OP_LESS:
           case SLAPI_OP_LESS_OR_EQUAL:
             lowerkey.dptr = slapi_ch_strdup(prefix);
@@ -1451,8 +1452,17 @@ index_range_read_ext(
     cur_key.data = lowerkey.data;
     cur_key.size = lowerkey.size;
     lowerkey.data = NULL; /* Don't need this any more, since the memory will be freed from cur_key */
-    if (operator == SLAPI_OP_GREATER) {
-        *err = index_range_next_key(db,&cur_key,db_txn);
+    *err = 0;
+    if (coreop == SLAPI_OP_GREATER) {
+        *err = index_range_next_key(db, &cur_key, db_txn);
+        if (*err) {
+            LDAPDebug(LDAP_DEBUG_ANY, "<= index_range_read(%s,%s) op==GREATER, no next key: %i)\n",
+                type, prefix, *err );
+            goto error;
+        }
+    }
+    if (operator & SLAPI_OP_RANGE_NO_ALLIDS) {
+        *err = NEW_IDL_NO_ALLID;
     }
     if (idl_get_idl_new()) { /* new idl */
         idl = idl_new_range_fetch(be, db, &cur_key, &upperkey, db_txn,
@@ -1462,7 +1472,7 @@ index_range_read_ext(
         int retry_count = 0;
         while (*err == 0 &&
                (upperkey.data &&
-                (operator == SLAPI_OP_LESS) ?
+                (coreop == SLAPI_OP_LESS) ?
                 DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) < 0 :
                 DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) <= 0)) {
             /* exit the loop when we either run off the end of the table,

ldap/servers/slapd/back-ldbm/init.c

@@ -36,7 +36,7 @@ ldbm_back_add_schema( Slapi_PBlock *pb )
 			SLAPI_ATTR_FLAG_NOUSERMOD );
 
 	rc |= slapi_add_internal_attr_syntax( LDBM_PARENTID_STR,
-			LDBM_PARENTID_OID, DIRSTRING_SYNTAX_OID, CASEIGNOREMATCH_NAME,
+			LDBM_PARENTID_OID, INTEGER_SYNTAX_OID, INTEGERMATCH_NAME,
 			SLAPI_ATTR_FLAG_SINGLE|SLAPI_ATTR_FLAG_NOUSERMOD );
 
 	rc |= slapi_add_internal_attr_syntax( "entryid",

ldap/servers/slapd/back-ldbm/ldbm_search.c

@@ -1200,6 +1200,8 @@ subtree_candidates(
     int isroot = 0;
     struct ldbminfo *li = (struct ldbminfo *) be->be_database->plg_private;
     int allidslimit = compute_allids_limit(pb, li);
+    Operation *op = NULL;
+    PRBool is_bulk_import = PR_FALSE;
 
     /* make (|(originalfilter)(objectclass=referral)) */
     ftop= create_subtree_filter(filter, managedsait, &focref, &forr);
@@ -1216,6 +1218,12 @@ subtree_candidates(
 
     has_tombstone_filter = (filter->f_flags & SLAPI_FILTER_TOMBSTONE);
     slapi_pblock_get( pb, SLAPI_REQUESTOR_ISROOT, &isroot );
+    /* Check if it is for bulk import. */
+    slapi_pblock_get(pb, SLAPI_OPERATION, &op);
+    if (entryrdn_get_switch() && operation_is_flag_set(op, OP_FLAG_INTERNAL) &&
+        operation_is_flag_set(op, OP_FLAG_BULK_IMPORT)) {
+        is_bulk_import = PR_TRUE;
+    }
 
     /*
      * Apply the DN components if the candidate list is greater than
@@ -1237,13 +1245,13 @@ subtree_candidates(
             candidates = idl_intersection(be, candidates, descendants);
             idl_free(&tmp);
             idl_free(&descendants);
-        } else if (!has_tombstone_filter) {
+        } else if (!has_tombstone_filter && !is_bulk_import) {
             *err = ldbm_ancestorid_read_ext(be, &txn, e->ep_id, &descendants, allidslimit);
             idl_insert(&descendants, e->ep_id);
             candidates = idl_intersection(be, candidates, descendants);
             idl_free(&tmp);
             idl_free(&descendants);
-        } /* else == has_tombstone_filter: do nothing */
+        } /* else == has_tombstone_filter OR is_bulk_import: do nothing */
     }
 
     return( candidates );
@@ -1683,13 +1691,17 @@ ldbm_back_next_search_entry_ext( Slapi_PBlock *pb, int use_extension )
            * >0 an ldap error code
            */
           int filter_test = -1;
+          int is_bulk_import = operation_is_flag_set(op, OP_FLAG_BULK_IMPORT);
         
-          if((slapi_entry_flag_is_set(e->ep_entry,SLAPI_ENTRY_LDAPSUBENTRY) 
-             && !filter_flag_is_set(filter,SLAPI_FILTER_LDAPSUBENTRY)) ||
-            (slapi_entry_flag_is_set(e->ep_entry,SLAPI_ENTRY_FLAG_TOMBSTONE)
-             && ((!isroot && !filter_flag_is_set(filter, SLAPI_FILTER_RUV)) ||
-             !filter_flag_is_set(filter, SLAPI_FILTER_TOMBSTONE))))
-          {
+          if (is_bulk_import) {
+              /* If it is from bulk import, no need to check. */
+              filter_test = 0;
+              slimit = -1; /* no sizelimit applied */
+          } else if ((slapi_entry_flag_is_set(e->ep_entry,SLAPI_ENTRY_LDAPSUBENTRY) &&
+                      !filter_flag_is_set(filter,SLAPI_FILTER_LDAPSUBENTRY)) ||
+                     (slapi_entry_flag_is_set(e->ep_entry,SLAPI_ENTRY_FLAG_TOMBSTONE) &&
+                      ((!isroot && !filter_flag_is_set(filter, SLAPI_FILTER_RUV)) ||
+                       !filter_flag_is_set(filter, SLAPI_FILTER_TOMBSTONE)))) {
             /* If the entry is an LDAP subentry and filter don't filter subentries OR 
              * the entry is a TombStone and filter don't filter Tombstone 
              * don't return the entry.  We make a special case to allow a non-root user
@@ -1736,7 +1748,8 @@ ldbm_back_next_search_entry_ext( Slapi_PBlock *pb, int use_extension )
             /* ugaston - if filter failed due to subentries or tombstones (filter_test=-1),
              * just forget about it, since we don't want to return anything at all. */
          {
-             if ( slapi_uniqueIDCompareString(target_uniqueid, e->ep_entry->e_uniqueid) ||
+             if ( is_bulk_import ||
+                  slapi_uniqueIDCompareString(target_uniqueid, e->ep_entry->e_uniqueid) ||
                   slapi_sdn_scope_test_ext( backentry_get_sdn(e), basesdn, scope, e->ep_entry->e_flags ))
              {
                  /* check size limit */
@@ -1781,7 +1794,7 @@ ldbm_back_next_search_entry_ext( Slapi_PBlock *pb, int use_extension )
                  }
                  rc = 0;
                  goto bail;
-             } 
+             }
              else 
              {
                  CACHE_RETURN ( &inst->inst_cache, &(sr->sr_entry) );

ldap/servers/slapd/back-ldbm/misc.c

@@ -79,6 +79,7 @@ static const char *systemIndexes[] = {
     SLAPI_ATTR_NSCP_ENTRYDN,
     ATTR_NSDS5_REPLCONFLICT,
     SLAPI_ATTR_ENTRYUSN,
+    SLAPI_ATTR_PARENTID,
     NULL
 };
 

ldap/servers/slapd/entry.c

@@ -3429,8 +3429,8 @@ slapi_entry_rename(Slapi_Entry *e, const char *newrdn, int deleteoldrdn, Slapi_D
 
     /* We remove the parentid and entrydn since the backend will change these.
      * We don't want to give the caller an inconsistent entry. */
-    slapi_entry_attr_delete(e, "parentid");
-    slapi_entry_attr_delete(e, "entrydn");
+    slapi_entry_attr_delete(e, SLAPI_ATTR_PARENTID);
+    slapi_entry_attr_delete(e, SLAPI_ATTR_ENTRYDN);
 
     /* Build new DN.  If newsuperior is set, just use "newrdn,newsuperior".  If
      * newsuperior is not set, need to add newrdn to old superior. */

ldap/servers/slapd/slap.h

@@ -536,12 +536,16 @@ typedef int (*SyntaxEnumFunc)(char **names, Slapi_PluginDesc *plugindesc,
 /* OIDs for some commonly used matching rules */
 #define DNMATCH_OID				"2.5.13.1"	/* distinguishedNameMatch */
 #define CASEIGNOREMATCH_OID		"2.5.13.2"	/* caseIgnoreMatch */
+#define INTEGERMATCH_OID		"2.5.13.14"	/* integerMatch */
+#define INTEGERORDERINGMATCH_OID		"2.5.13.15"	/* integerOrderingMatch */
 #define INTFIRSTCOMPMATCH_OID	"2.5.13.29"	/* integerFirstComponentMatch */
 #define OIDFIRSTCOMPMATCH_OID	"2.5.13.30"	/* objectIdentifierFirstComponentMatch */
 
 /* Names for some commonly used matching rules */
 #define DNMATCH_NAME			"distinguishedNameMatch"
 #define CASEIGNOREMATCH_NAME	"caseIgnoreMatch"
+#define INTEGERMATCH_NAME		"integerMatch"
+#define INTEGERORDERINGMATCH_NAME		"integerOrderingMatch"
 #define INTFIRSTCOMPMATCH_NAME	"integerFirstComponentMatch"
 #define OIDFIRSTCOMPMATCH_NAME	"objectIdentifierFirstComponentMatch"
 

ldap/servers/slapd/slapi-plugin.h

@@ -483,6 +483,7 @@ NSPR_API(PRUint32) PR_fprintf(struct PRFileDesc* fd, const char *fmt, ...)
 #define SLAPI_ATTR_ENTRYDN 			"entrydn"
 #define SLAPI_ATTR_DN 				"dn"
 #define SLAPI_ATTR_RDN 				"rdn"
+#define SLAPI_ATTR_PARENTID			"parentid"
 #define SLAPI_ATTR_UNIQUEID_LENGTH		10
 #define SLAPI_ATTR_OBJECTCLASS_LENGTH		11
 #define SLAPI_ATTR_VALUE_TOMBSTONE_LENGTH	11
@@ -494,6 +495,7 @@ NSPR_API(PRUint32) PR_fprintf(struct PRFileDesc* fd, const char *fmt, ...)
 #define SLAPI_ATTR_ENTRYDN_LENGTH 		7
 #define SLAPI_ATTR_DN_LENGTH 			2
 #define SLAPI_ATTR_RDN_LENGTH 			3
+#define SLAPI_ATTR_PARENTID_LENGTH 		8
 
 /* plugin shared config area */
 #define SLAPI_PLUGIN_SHARED_CONFIG_AREA "nsslapd-pluginConfigArea"
@@ -7024,6 +7026,9 @@ typedef struct slapi_plugindesc {
 #define SLAPI_OP_GREATER_OR_EQUAL			4
 #define SLAPI_OP_GREATER				5
 #define SLAPI_OP_SUBSTRING				6
+#define SLAPI_OP_RANGE					0xff
+#define SLAPI_OP_RANGE_NO_IDL_SORT		0x100
+#define SLAPI_OP_RANGE_NO_ALLIDS		0x200
 
 /* Defined values of SLAPI_PLUGIN_MR_USAGE: */
 #define SLAPI_PLUGIN_MR_USAGE_INDEX		0
@@ -7607,7 +7612,8 @@ enum
     BACK_INFO_CRYPT_ENCRYPT_VALUE, /* Ctrl: clcrypt_encrypt_value */
     BACK_INFO_CRYPT_DECRYPT_VALUE, /* Ctrl: clcrypt_decrypt_value */
     BACK_INFO_DIRECTORY,           /* Get the directory path */
-    BACK_INFO_LOG_DIRECTORY        /* Get the txn log directory */
+    BACK_INFO_LOG_DIRECTORY,       /* Get the txn log directory */
+    BACK_INFO_IS_ENTRYRDN          /* Get the flag for entryrdn */
 };
 
 struct _back_info_crypt_init {

ldap/servers/slapd/slapi-private.h

@@ -416,6 +416,7 @@ char *slapi_filter_to_string_internal( const struct slapi_filter *f, char *buf,
 #define OP_FLAG_REVERSE_CANDIDATE_ORDER  0x100000 /* reverse the search candidate list */
 #define OP_FLAG_NEVER_CACHE              0x200000 /* never keep the entry in cache */
 #define OP_FLAG_TOMBSTONE_FIXUP          0x400000 /* operation is tombstone fixup op */
+#define OP_FLAG_BULK_IMPORT              0x800000 /* operation is bulk import */
 
 /* reverse search states */
 #define REV_STARTED 1