# ImportWorker.java

    /*
    ** Module   : ImportWorker.java
    ** Abstract : Pattern worker used to import Progress data into a relational database using batch insert.
    **
    ** Copyright (c) 2004-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- -----------------------------------Description-----------------------------------
    ** 001 ECF 20050715   @21822 Created initial version. Includes support to insert new records in
    **                           batch mode, and to query individual records in existing tables
    **                           to resolve foreign associations.
    ** 002 ECF 20050801   @21970 Substantial rewrite to add fault tolerance and improve performance.
    **                           Much of the function previously implemented in the import.xml
    **                           ruleset has been moved into the importTable user function.
    ** 003 ECF 20050805   @22014 Minor changes to error logging and progress indicator.
    **                           Also some doc additions.
    ** 004 ECF 20050809   @22078 Added analyze step to end of table import.
    **                           Runs the postgres-specific "analyze" command.
    **                           This eventually will need to be made vendor neutral.
    ** 005 ECF 20050810   @22081 Modified QueryHelper and calling code to fix parameter name mismatch.
    **                           In some cases, pairs of properties involved in a join of two classes
    **                           can have their names fall out of synch during name conversion.
    **                           Modified the parameter management in QueryHelper to handle this
    **                           disconnect. Also introduced a separate session for foreign
    **                           association queries only, which is never flushed and cleared much
    **                           less often. This should help query performance, as objects can be
    **                           retrieved from the first level cache more often.
    ** 006 ECF 20050819   @22205 Removed RecordLoader and PropertyMapper inner classes (and all
    **                           subclasses). These classes were abstracted into their own top level
    **                           classes to support sharing between database and directory import
    **                           tasks. Created inner class SqlRecordLoader which extends the new
    **                           RecordLoader class to handle database import.
    ** 007 ECF 20050930   @22962 Added dynamic creation of indexes during table import. This was
    **                           necessary to work around a Hibernate limitation which does not
    **                           guarantee the order of index columns during normal schema DDL
    **                           generation. Also, by doing it this way, we avoid the performance
    **                           hit of updating all non-unique indexes for each insert. Instead, we
    **                           build only unique indexes up front (to ensure data integrity), then
    **                           defer creation of the remaining indexes until after the data has
    **                           been imported. Also added use of Hibernate's query cache to speed up
    **                           queries against tables which have one-to-many associations with
    **                           tables being imported.
    ** 008 ECF 20051027   @23201 Added support for case-insensitive index components. Use the database
    **                           upper() function to ignore case when creating indexes on case-
    **                           insensitive text columns.
    ** 009 ECF 20051103   @23216 Removed user functions. nonRedundantIndexes
    **                           and normalizeUniqueIndexes were removed.
    ** 010 ECF 20060713   @28039 Made improvements in indexing support. Added reindexing feature,
    **                           which drops and recreates all indexes defined for each database
    **                           table.
    ** 011 ECF 20060714   @28137 Updated QueryHelper inner class. Changes were necessary to emit HQL
    **                           using the new RTRIM syntax embedded for character properties.
    ** 012 ECF 20070312   @32389 Multi-threaded the import worker. Each table is imported in a
    **                           dedicated thread. The number of threads which may run simultaneously
    **                           is configurable. This change significantly improves the throughput
    **                           of the import job on multi-processor hardware. There should be
    **                           some improvement on a uni-processor system as well, though it will
    **                           not be as dramatic.
    ** 013 ECF 20070314   @32411 Reworked import job prioritization. Jobs are now ranked and sorted
    **                           to ensure that critical import paths are optimized for maximum
    **                           throughput. Also reworked job dispatching model to proactively
    **                           schedule jobs, rather than leaving this up to the JVM's thread
    **                           scheduler.
    ** 014 ECF 20070424   @33214 Removed text-only optimization in reindex(). In some cases,
    **                           re-establishing all indices for a table is necessary.
    ** 015 EVL 20070609   @34005 Adding explicit import of the class org.hibernate.type.Type to
    **                           eliminate conflict with the same class from java.lang.reflect
    **                           package to be able to compile for Java 6.
    ** 016 ECF 20070705   @34362 Minor optimization. Replaced StringBuffer with StringBuilder.
    ** 017 ECF 20070908   @35340 Fixed memory problem. Use SymbolResolver constructor which
    **                           disables SchemaDictionary. We don't need schema lookups
    **                           and that class is not thread-safe, so we were creating a
    **                           huge memory footprint by creating multiple SchemaDictionary
    **                           instances across threads. Replaced record counter ints with longs.
    ** 018 ECF 20080104   @36679 Set ErrorManager to headless mode. Better error reporting for
    **                           errors encountered while loading next record from export data.
    ** 019 ECF 20080306   @37467 Modified to support changes in DBUtils.
    ** 020 GES 20080418   @38172 Moved from a lexer approach to using the equivalent runtime support
    **                           for the Progress IMPORT language stmt. This allows the import
    **                           process to match the behavior of Progress which seems to internally
    **                           use the IMPORT stmt. Note that this allows proper date component
    **                           ordering, non-ASCII character set and other I18N support to
    **                           work properly.
    ** 021 ECF 20080804   @39292 Reworked primary key generation strategy. All primary keys generated
    **                           for a given database must be unique across all tables.
    ** 022 ECF 20081124   @40686 Remove analyze command after table import. This is no longer
    **                           necessary, since we are not immediately running queries for foreign
    **                           key lookups against newly imported tables. On older versions of
    **                           PostgreSQL, this was causing conflicts with autovacuum.
    ** 023 GES 20090422   @41898 Converted to standard string formatting.
    ** 024 GES 20100630          Refactored code to expose a new method called generateIndexDDL().
    **                           This allows the DDL which is normally created during data import
    **                           to be captured to an output stream. Cleaned up some unnecessary
    **                           usage of database metadata during initialize().
    ** 025 CA  20101026          Moved the CREATE/DROP INDEX statements to be generated by the
    **                           dialect. Set the default value for any not-null property which has
    **                           missing data.
    ** 026 OM  20121210          Added import sequence value.
    ** 027 ECF 20121212          Moved setting of date-order from runImport to importAsync. This
    **                           setting is context-local, so it has to be set on the job thread.
    **                           Skip word index in {create|drop}Indexes methods. Fixed import of
    **                           sequence values to consider name conversion.
    **                           Fixed import of lone hyphen entry from an export
    **                           file (was being skipped).
    ** 028 SVL 20130331          Upgraded to Hibernate 4.
    ** 029 ECF 20130510          Fixed sequence initialization, which would error out for some
    **                           database dialects.
    ** 030 OM  20130516          Added mappings for int64, datetime/datetime-tz expressions
    **                           in QueryHelper class.
    ** 031 OM  20130527          Remapped DatetimeTzUserType as a CompositeCustomType.
    **                           For the moment this is disabled (1584#150).
    **                           Replaced deprecated FlushMode.NEVER field
    **                           Added PSC footer handling.
    ** 032 CA  20130822          The DATE-FORMAT attribute must be accessed via SessionUtils.
    ** 033 VMN 20131005          Added check hasFullSequenceSupport for using sequences in
    *                            generateIndexDDL.
    ** 034 VMN 20131106          Removed check hasFullSequenceSupport: we will assume all supported
    **                           databases at least can create p2j_id_generator_sequence during DDL
    **                           generation.
    ** 035 OM  20130923          The passwords for _User meta-table are read form dedicated file.
    **                           Switched logging to java.util.logging API instead of apache's.
    ** 036 ECF 20140107          Fixed regression which wasn't committing newly created indexes.
    ** 037 OM  20140410          Fixed support case-sensitive fields in indexes.
    ** 038 OM  20140625          Use the new signature of method DBUtils.composePropertyName.
    ** 039 ECF 20141021          Minor source code cleanup.
    ** 040 OM  20141111          Refactored setDateOrder() to setDateFormat() in SessionUtils.
    ** 041 OM  20150318          Removed support for plain test password files because now we have a
    **                           P4GL fully compatible encode function. Upgraded code to Java7.
    ** 042 OM  20151217          Replaced 'runner' anonymous class with lambda in importAsync().
    ** 043 OM  20160119          Fixed removeFailingRecord() method.
    ** 044 EVL 20160224          Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
    ** 045 OM  20160603          Fixed rtrimming to space character only.
    ** 046 OM  20160712          Removed p2j_id_generator_sequence from generated index DDL.
    ** 047 HC  20170425          Explicitly disabled bean validation for Hibernate in case a
    **                           JSR 303 validation appears on the classpath (GWT for example).
    ** 048 ECF 20181214          Added LOB support.
    ** 049 ECF 20190309          Fixed recovery for failed statement(s) in a batch update. The info on
    **                           update counts provided by BatchUpdateException is no longer reliable.
    ** 050 CA  20190905          The data file is not mapped to memory, but to a byte buffer.  This
    **                           allows very large dump files to be imported, without limitations from
    **                           the physical memory.
    ** 051 OM  20191120          Transaction.rollback throws exception.
    **     OM  20200507          Upgraded to new persistence framework, without Hibernate.
    ** 052 OM  20200924          P2JIndexComponent carries multiple information to avoid map lookups for them.
    ** 053 IAS 20201219          Added 'CPSTREAM' use on import.
    **                           Added word tables population on import.
    **     IAS 20210219          Word tables support for custom extents.
    **     OM  20210525          Fixed support for processing really big dump files. Enabled multibyte encoding.
    **                           Use the correct character encoding when processing dump files (read from footer).
    **     OM  20210616          Added heuristic method for locating the PCS footer. Unified logging messages.
    **                           Synchronized logs by sending all messages to System.out stream.
    **     OM  20210702          Added support for mapping invalid bytes detected during reading of input files.
    **     CA  20210709          Create the identity sequence even if there are no data files to import.
    **                           Refactored to explicitly read bytes and not characters (and decode them as
    **                           needed), when resolving the PSC footer.
    **     OM  20210825          Improving messages logged during import.
    **     OM  20210831          Synchronized accesses to idContextStack.
    **     CA  20210920          After decoding, the CharBuffer needs to be used via 'toString()', and not via
    **                           'array()', as that returns the internal array which may have additional elements.
    **     OM  20211215          p2j_id_generator_sequence sequence is created even if fatal errors were recorded.
    **     OM  20220317          More p2j_id_generator_sequence awareness. If the sequence is detected, it is used
    **                           as base counter for imported records. At the end, its current value is updated.
    **     OM  20220405          Database artifacts are loaded from the application's jar.
    **                           Got rid of psql exception when updating the p2j_id_generator_sequence sequence.
    **     OM  20220428          Avoided potential RuntimeException when the id sequence was not defined in psql.
    **     OM  20220602          Unified parsing of invalid bytes mapping string with DatabaseDumpChecker.
    **     SBI 20221023          Added a usage of SQLQuery.uniqueResult(Session session).
    **     RAA 20221221          Modified the execution of sql's. They now pass through the SQLStatementLogger
    **                           in case logging is intended.
    **     OM  20220910          Fixed reading and processing of empty records.
    **     OM  20221007          Wasted a value in ID_GEN_SEQUENCE's initialization because some dialects have
    **                           issues when requesting the CurVal if positioned at start.
    **     OM  20221013          Insert a file-path separator in sequence dump file path if one is needed.
    **     IAS 20221125          Make location of LOB files configurable for import
    **     SVL 20230110          P2JIndex.components() provides direct access to the components array in order to
    **                           improve performance.
    **     SBI 20221023          Added a usage of SQLQuery.uniqueResult(Session session).
    **     IAS 20230222          Fixed word tables import to support denorm-extents=true.
    ** 054 HC  20230427          Implemented server-side file-system resource. File operations
    **                           performed on client can be optionally performed directly on server
    **                           without the expensive remote call.
    ** 055 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 056 RAA 20230518          Replaced lambdas with method referencing when using SQLStatementLogger.
    ** 057 RAA 20230607          SQLs are now executed through SQLExecutor instead of SQLStatementLogger.
    ** 058 OM  20231016          Improved logging support.
    **     RFB 20231120          Fixed minor typo in a logging line.
    **     OM  20231213          Skipped imports to word tables if current dialect does not support them.
    ** 059 TJD 20240123          Java 17 compatibility updates
    ** 060 SP  20240619          Set character encoding to use in Functions before imports to word tables.
    **     SP  20240621          Added a new method wordReindexAsync() to re-populate the word tables.
    ** 061 SP  20240704          Replaced CALL with SELECT in wordReindexAsync for calling the UDF.
    ** 062 DDF 20231025          Added empty flag which allows creating an empty database.
    **     DDF 20231026          Always import meta_user table records, even an empty database is currently
    **                           being created.
    **     DDF 20250218          Make use of the mandatory tables from MetadataManager to determine what
    **                           tables should be imported for an empty import.
    **     DDF 20250218          Added legacyName parameter to SqlRecordLoader().
    **     DDF 20250219          Replace usage of mandatory tables with the regular tables.
    */

    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** This program is distributed in the hope that it will be useful,
    ** but WITHOUT ANY WARRANTY; without even the implied warranty of
    ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    ** GNU Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.schema;

    import java.beans.*;
    import java.io.*;
    import java.nio.*;
    import java.nio.charset.*;
    import java.nio.file.*;
    import java.sql.*;
    import java.sql.Statement;
    import java.util.*;
    import java.util.logging.*;
    import java.util.stream.*;

    import org.apache.commons.lang3.*;
    import com.goldencode.p2j.cfg.*;
    import com.goldencode.p2j.convert.*;
    import com.goldencode.p2j.pattern.*;
    import com.goldencode.p2j.persist.*;
    import com.goldencode.p2j.persist.Record;
    import com.goldencode.p2j.persist.dialect.*;
    import com.goldencode.p2j.persist.meta.MetadataManager.SystemTable;
    import com.goldencode.p2j.persist.orm.*;
    import com.goldencode.p2j.persist.orm.types.*;
    import com.goldencode.p2j.persist.pl.Functions;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.ErrorManager;
    import com.goldencode.util.*;
    import com.mchange.v2.c3p0.*;

    /**
     * A pattern worker that provides batch data import services. Exposes user functions which assist in inserting
     * data read from Progress {@code *.d} data dump files into a relational database.  Defines a number of inner,
     * helper classes, most of which are configured before a table is imported.
     * <p>
     * This worker enables a multi-threaded import, where different tables are
     * loaded in parallel threads.  The number of threads which may actively be
     * importing data at once is set with the {@link
     * ImportWorker.Library#setMaximumThreads setMaximumThreads} method, which is
     * exposed to TRPL as a user function.
     * <p>
     * The idiom used to run an import bundle is to walk through the P2O tree of the
     * target database using the <code>schema/import.xml</code> ruleset.  This
     * ruleset prepares the necessary {@link ImportWorker.SqlRecordLoader}
     * instances and lists of indexes which must be created for each table.  It
     * then invokes the {@link ImportWorker.Library#prepareImport prepareImport}
     * user function for each table.  This creates the necessary data structures
     * in preparation for the import.
     * <p>
     * When all tables have been prepared, the import ruleset invokes the {@link
     * ImportWorker.Library#runImport runImport} user function.  This analyzes
     * the tables targeted for import, their associated data export files, and
     * their dependencies and dependent tables to determine the critical paths
     * which must be traversed during the import.  The tables are then sorted
     * into priority order to ensure that all critical paths are traversed as
     * directly as possible.
     * <p>
     * Once tables have been prioritized and sorted, the import begins.  The
     * prioritized queue of import bundles is traversed in order, and a thread is
     * launched for each import bundle in the queue, up to the maximum number of
     * active threads.  When a thread completes importing a table, the next
     * available bundle is removed from the queue and processed.  A bundle is only
     * allowed to commence if all tables upon which it is dependent have been
     * imported first.  If this is not the case, this bundle is skipped and the next
     * waiting bundle is tested.  This continues until a bundle is found which can
     * commence immediately, or until the end of the bundle queue is reached.  If
     * no available bundle was found, but the bundle queue is not empty, the main
     * thread waits until one of the active threads completes its bundle, then runs
     * through the queue again, looking for an available candidate.  This process
     * continues until the bundle queue is empty.
     * <p>
     * Once all active import threads have completed and no more bundles are waiting
     * in the bundle queue, the import run is complete and the <code>runImport</code>
     * method returns control to TRPL.
     * <p>
     * Progress statistics are maintained and reported at the completion of each
     * bundle in the queue.  These include percent complete (based upon export file
     * bytes processed), records loaded, and average number of records processed
     * across various periods.  Because these are only calculated and reported
     * upon completion of a bundle, these statistics trend lower than the actual
     * throughput being achieved.  This is because at the point where one thread
     * completes its work for a single table, all other threads are at some state
     * of partial completion of their own bundles, so the actual throughput is not
     * accurately reflected in the current calculations.  However, the trade-off
     * for more accurate, real-time statistics would have been a higher level of
     * synchronization and a more complicated algorithm to account for dropped
     * records and actual records imported.  Rather than possibly compromise
     * throughput with such an implementation, it was deemed better to settle for
     * less accurate statistics.  The final set of statistics reported at the end
     * of the full run is reflective of the actual, average performance achieved.
     * <p>
     * Moving from a single-threaded (as this worker originally was designed) to
     * a multi-threaded architecture paid huge dividends, particularly on
     * multi-processor hardware.  On a 4-processor, dual-core system, running
     * with 8 active threads, an import of approximately 79 million primary
     * records completed in approximately 40% of the time needed for a
     * single-threaded run on the same system.  Gains on a uni-processor system
     * are not expected to be as dramatic, though some slight improvement is
     * likely.  Performance tends to scale directly with the number of (virtual)
     * processors on a system.  It is recommended to run with one thread per
     * virtual processor for best results.
     */
    public final class ImportWorker
    extends AbstractPatternWorker
    {
       /** Logger. */
       private static final ConversionStatus LOG = ConversionStatus.get(ImportWorker.class);

       /** The cached name converter */
       private static final NameConverter NC = new NameConverter();

       /** Directory path containing export files for the database being imported */
       static String dataPath = null;

       /** Directory path containing export LOB data for the database being imported */
       static String lobPath = null;

       /** The database we import into. */
       static Database database = null;

       /** The ORM settings, including C3P0 and database connection. */
       static Settings ormSettings = null;

       /** The data source used to connect to database when necessary. */
       static ComboPooledDataSource cpds = null;

       /**
        * Value representing if records should be imported. If set to <code>true</code>,
        * records will not be imported, but indexes and sequences will be created.
        * It is set to <code>false</code> by default.
        */
       static boolean empty = false;

       /**
        * The default {@code numformat} attribute for data ({@code .d}) files which lack it from the PSC footer or
        * the PSC footer completely. Ex: "46,44".
        */
       private static String defaultNumFormat = null;

       /**
        * The default {@code dateformat} attribute for data ({@code .d}) files which lack it from the PSC footer
        * or the PSC footer completely. Ex: "dmy-1950".
        */
       private static String defaultDateFormat = null;

       /**
        * The default {@code cpstream} attribute for data ({@code .d}) files which lack it from the PSC footer or
        * the PSC footer completely. Ex: "1252".
        */
       private static String defaultCpStream =  null;

       /**
        * The mapped bytes used in an attempt to fix broken encoding in dump files. If present, these bytes will
        * be mapped (replaced with the value) only if the decoding {@code Charset} will detect an issue with it.
        */
       private static Byte[] mapBytes = null;

       /** Default JDBC batch size if not in database configuration. */
       private static final int DEFAULT_BATCH_SIZE = 50;

       /** Format specifier for an import statistics update message. */
       private static final String STATS_FMT =
             "%.5f%% complete:  %d recs in %d:%02d:%06.3f (%d/sec, %d/min, %d/hr)%n";

       /** Map of unqualified DMO names to import bundles */
       private final Map<String, ImportBundle> importBundles = new LinkedHashMap<>();

       /** Map of historical to SQL sequence names */
       private final Map<String, String> sequences = new HashMap<>();

       /** Set of unqualified DMO class names for tables which were processed */
       private final Set<String> complete = new HashSet<>();

       /** Set of unqualified DMO class names for tables which had fatal errors */
       private final Set<String> fatal = new HashSet<>();

       /** Set of all DMO classes for which an import was attempted */
       private final Set<String> attempted = new HashSet<>();

       /** Lock for record count synchronization */
       private final Object countLock = new Object();

       /**
        * Deque of IDContexts, up to {@link #maximumThreads} deep. This field is final to support the
        * synchronizing statement from {@link IDContext#prepareNextBracket()} method.
        */
       private final Deque<IDContext> idContextStack = new ArrayDeque<>();

       /** Queue of import bundles */
       private List<ImportBundle> bundleQueue = null;

       /** JDBC batch size */
       private int batchSize = 0;

       /** Database dialect in use */
       private Dialect dialect = null;

       /** Number of import threads allowed to be active simultaneously */
       private int maximumThreads = 4;

       /** Number of import threads currently active */
       private int activeThreads = 0;

       /** Number of records imported */
       private long imported = 0;

       /** Stores the number of errors for each table individually. */
       private final Map<String, Integer> errors = new LinkedHashMap<>();

       /** Number of bytes imported */
       private double importedBytes = 0.0;

       /** Total byte count of all exported data files */
       private double totalBytes = 0.0;

       /** System time at beginning of import */
       private long startTime = 0L;

       /** Next ID bracket */
       private long nextGlobalID = 0L;

       /** ID bracket size */
       private final long idBracketSize = 10000L;

       /** Do we create the {@code p2j_id_generator_sequence} or update it? Default is update.*/
       private boolean createGenerator = false;

       /**
        * Default constructor which initializes callback library.
        */
       public ImportWorker()
       {
          super();

           // Entering import mode. This method must be called from the @import.xml@ as soon as possible in order
           // to allow database artifacts to be read from the application jar instead of conversion environment.
          Configuration.setImportMode();

          setLibrary(new Library());
          ErrorManager.setHeadless();
       }

       /**
        * Open a new database session and set it to flush only on explicit flushes.
        *
        * @return  New session.
        */
       private Session openSession()
       throws PersistenceException
       {
          if (cpds == null)
          {
             cpds = new ComboPooledDataSource();

             Dialect dialect = Dialect.getDialect(ormSettings);
             String driver = dialect.getDriverClassName();
             try
             {
                cpds.setDriverClass(driver);
             }
             catch (PropertyVetoException pve)
             {
                throw new PersistenceException(pve);
             }

             // basic connection settings
             String url = ormSettings.getString(Settings.URL, null);
             cpds.setJdbcUrl(url);
             String user = ormSettings.getString(Settings.USER, null);
             cpds.setUser(user);
             String pass = ormSettings.getString(Settings.PASSWORD, null);
             cpds.setPassword(pass);

             // additional configuration
             for (Map.Entry<String, Object> next : ormSettings.getAllForCategory("c3p0").entrySet())
             {
                String key = next.getKey();
                String value = String.valueOf(next.getValue());
                System.setProperty(key, value);
             }

             // setup the blob and clob creators
             TypeManager.addBlobCreator(url, dialect::blobCreator);
             TypeManager.addClobCreator(url, dialect::clobCreator);
          }

          return new Session(database, cpds);
       }

       /**
        * Determines the high-water primary key generated during the import and creates an identity
        * generator sequence whose starting value is that ID. Presumes backing database supports
        * sequences.
        */
       private void createIdentitySequence()
       {
          if (imported == 0 && !createGenerator)
          {
             return; // no-op: generator sequence already exists and no records were processed
          }

          long highID = nextGlobalID;
          synchronized (idContextStack)
          {
             for (IDContext idCtx : idContextStack)
             {
                highID = Math.max(highID, idCtx.getCurrentID());
             }
          }

          String ddl = createGenerator ? dialect.getCreateSequenceString(Persistence.ID_GEN_SEQUENCE, highID)
                                       : dialect.getSequenceSetValString(Persistence.ID_GEN_SEQUENCE, highID);

          if (ddl == null)
          {
             System.out.println("Creation of identity generator sequence starting at " +
                                highID + " is not supported.");
             return;
          }

          Session session = null;
          boolean inTx = false;
          boolean expectsResult = ddl.toLowerCase().startsWith("select");

          try
          {
             session = openSession();
             inTx = session.beginTransaction();
             if (expectsResult)
             {
                Session.createSQLQuery(ddl).uniqueResult(session); // ignore result
             }
             else
             {
                Session.createSQLQuery(ddl).executeUpdate(session);
             }
             session.commit();

             System.out.println((createGenerator ? "Created" : "Updated") +
                                " identity generator sequence starting at " + highID + ".");
          }
          catch (PersistenceException exc)
          {
             if (inTx)
             {
                try
                {
                   session.rollback();
                }
                catch (PersistenceException rbEx)
                {
                   LOG.log(Level.SEVERE, "", rbEx);
                }
             }

             // do not log message if the sequence already existed. This is expected in case of
             // multi-session imports
             if (exc.getMessage().contains("already exists"))
             {
                // TODO: maybe update it with the new [highID] value
                System.out.println("Detected the " + Persistence.ID_GEN_SEQUENCE + " sequence already created.");
             }
             else
             {
                LOG.log(Level.INFO, "Failed to create/update identity sequence", exc);
             }
          }
          finally
          {
             if (session != null && session.isOpen())
             {
                try
                {
                   session.close();
                }
                catch (PersistenceException exc)
                {
                   LOG.log(Level.FINE, "Failed to close session (create/update of " + Persistence.ID_GEN_SEQUENCE + ")", exc);
                }
             }
          }
       }

       /**
        * Open a new buffered output stream for the given target filename.
        *
        * @param    filename
        *           Target output file.  Must not reference a directory or any
        *           resource that cannot be created or written to.
        *
        * @return   The newly opened stream.
        *
        * @throws   IllegalArgumentException
        *           If the file cannot be properly accessed or opened.
        */
       private OutputStream openStream(String filename)
       {
          File file = new File(filename);

          if (file.isDirectory())
          {
             throw new IllegalArgumentException(filename + " is a directory!");
          }

          try
          {
             // this is a file, get the containing directory
             File parent = file.getParentFile();

             if (parent != null && !parent.exists())
             {
                // the directory doesn't exist, so create it (including all intermediate directories
                // needed
                if (!parent.mkdirs())
                {
                   throw new IllegalArgumentException(
                         "Failed to create " + filename + " directory! No write right?");
                }
             }

             // create/overwrite the file
             FileOutputStream fos = new FileOutputStream(file, true);

             // init our buffered stream wrapper
             return new BufferedOutputStream(fos);
          }
          catch (Exception exc)
          {
             throw new IllegalArgumentException("Cannot create stream for " + filename + "(" + exc + ")!", exc);
          }
       }

       /**
        * Output the given text to the stream in UTF-8 encoding.
        *
        * @param    os
        *           Output destination.
        * @param    text
        *           Content to write.
        */
       private void write(OutputStream os, String text)
       throws IOException
       {
          os.write(text.getBytes(StandardCharsets.UTF_8));
       }

       /**
        * Output the given text as a statement (delimited and with a new line) to the stream in UTF-8 encoding.
        *
        * @param    os
        *           Output destination.
        * @param    text
        *           Content to write.
        * @param    delimiter
        *           Statement ending delimiter.
        */
       private void writeStmt(OutputStream os, String text, String delimiter)
       throws IOException
       {
          write(os, text + delimiter + "\n");
       }

       /**
        * Library of callback user functions exposed to rulesets.
        */
       public class Library
       {
          /**
           * Initializes the object using the information specified in the given {@code Properties} and build the
           * resulting session factory. One particular parameter, JDBC batch size, is explicitly configured to a
           * default value if it is not specified in the configuration file.
           * <p>
           * In addition, the names of all text columns are mapped by table name for use during
           * index creation.
           *
           * @param   props
           *          The import properties.
           * @param   dataPath
           *          Path of directory containing data export files, and parent directory of LOB
           *          export directory, if any.
           * @param   lobPath
           *          Path of directory containing export LOB data.
           * @param   emptyImport
           *          Value representing the type of import. If <code>true</code>, records will not
           *          be imported.
           */
          public void initialize(Properties props, String dataPath, String lobPath, boolean emptyImport)
          {
             ImportWorker.dataPath = dataPath;
             ImportWorker.lobPath = lobPath;
             ImportWorker.empty = emptyImport;

             String databaseName = props.getProperty(Settings.DATABASE_NAME);
             ImportWorker.database = new Database(databaseName, Database.Type.PRIMARY, true);

             // detects and save off dialect for later use
             ImportWorker.ormSettings = new Settings(database, props);
             dialect = Dialect.getDialect(ormSettings);

             // ensure batch size is set to a reasonable value for import
             batchSize = props.containsKey(Settings.BATCH_SIZE)
                         ? (int) props.get(Settings.BATCH_SIZE)
                         : 0;
             if (batchSize == 0)
             {
                batchSize = DEFAULT_BATCH_SIZE;
                System.out.println("INFO: Using default batch size of " + batchSize + " for import.");
                ormSettings.put(Settings.BATCH_SIZE, batchSize);
             }

             // read the default values for [numformat], [dateformat] and [cpstream]
             SchemaConfig config = Configuration.getSchemaConfig();
             defaultNumFormat = config.getNamespaceParameter(databaseName, "numformat");
             defaultDateFormat = config.getNamespaceParameter(databaseName, "dateformat");
             defaultCpStream = config.getNamespaceParameter(databaseName, "cpstream");

             // extract the mapped bytes, if present
             mapBytes = DatabaseDumpChecker.parseByteMapping(
                   config.getNamespaceParameter(databaseName, "byte-mapping"));

             startTime = System.currentTimeMillis();
          }

          /** Check is file path is absolute.
           *
           * @param path
           *        file path to be checked
           * @return <code>true</code> is file path is absolute.
           */
          public boolean isAbsolute(String path)
          {
             return Paths.get(path).isAbsolute();
          }
          /**
           * Open a new database session.
           *
           * @return  New session.
           *
           * @throws  PersistenceException
           *          If any issue is encountered during the operation.
           */
          public Session openDatabaseSession()
          throws PersistenceException
          {
             return ImportWorker.this.openSession();
          }

          /**
           * Executes a query statement that returns an unique result and return the respective value.
           *
           * @param   session
           *          The session that holds the connection to the database to be used.
           * @param   queryText
           *          The SQL query statement.
           *
           * @return  The result of the query statement as a single  and simple object.
           *
           * @throws  PersistenceException
           *          If any issue is encountered during the operation.
           */
          public Object executeUniqueResultSqlQuery(Session session, String queryText)
          throws PersistenceException
          {
             return Session.createSQLQuery(queryText).uniqueResult(session);
          }

          /**
           * Set the maximum allowable threads.  This sets the number of threads
           * which may run table imports simultaneously.  However, a larger
           * number of threads may be created.  Additional threads may block
           * waiting for table upon which they are dependent to complete their
           * respective imports.
           *
           * @param   maximumThreads
           *          Maximum number of import threads which may be active at one
           *          time.
           */
          public void setMaximumThreads(int maximumThreads)
          {
             synchronized (complete)
             {
                ImportWorker.this.maximumThreads = maximumThreads;
             }
          }

          /**
           * Store the given objects in preparation for an import run.  The
           * import for all prepared tables commences, potentially on multiple
           * threads, when the {@link #runImport} method is invoked.
           *
           * @param   loader
           *          Record loader helper which contains all information
           *          necessary to find the import data and to map each record
           *          in the import file to a DMO of the appropriate class.
           * @param   indexes
           *          A list of index definitions which must be applied to the
           *          table before (for unique indexes) and after (for non-unique
           *          indexes) data import.
           *
           * @see     #setMaximumThreads
           */
          public void prepareImport(SqlRecordLoader loader, List<P2JIndex> indexes)
          {
             importBundles.put(loader.getDMOName(), new ImportBundle(loader, indexes));
          }

          /**
           * Commence an import run, potentially on multiple threads.  All tables
           * prepared by {@link #prepareImport} will be imported.
           * <p>
           * This method organizes the tables into an order which is intended to
           * maximize throughput throughout the import run, by scheduling those
           * tables with the most time-consuming import path to run earliest in
           * the process.
           * <p>
           * Multiple, parallel threads are used to import multiple tables at once.
           * <p>
           * This method will block until all prepared tables have been imported.
           *
           * @return  The number of records successfully imported.
           *
           * @throws  InterruptedException
           *          if the thread on which this method is invoked is interrupted.
           *
           * @see     #prepareImport
           * @see     #setMaximumThreads
           */
          public long runImport()
          throws InterruptedException
          {
             if (importBundles.isEmpty())
             {
                System.out.println("*** Nothing to import!");
                createIdentitySequence();
                return 0;
             }

             // setup JVM-wide configuration values to support certain runtime behavior that would
             // otherwise come from the directory.
             String dorder = Configuration.getParameter("date-order");
             String wyear  = Configuration.getParameter("windowing-year");
             Long yearOffset = null;
             try
             {
                yearOffset = Long.parseLong(wyear);
             }
             catch (NumberFormatException e)
             {
                // on error offsetYear stays null and will be handled later in the new thread
             }

             String gsep = Configuration.getParameter("number-group-sep");
             String dsep = Configuration.getParameter("number-decimal-sep");
             if (gsep != null && gsep.length() == 1 && dsep != null && dsep.length() == 1)
             {
                NumberType.overrideDefaultSeparators(gsep.charAt(0),dsep.charAt(0));
             }

             // check whether the [p2j_id_generator_sequence] is active and initialize [nextGlobalID] from it
             Session session = null;
             try
             {
                session = openSession();
                session.beginTransaction();
                String sql = dialect.getSequenceNextValString(Persistence.ID_GEN_SEQUENCE);
                Object result = null;
                try
                {
                   result = Session.createSQLQuery(sql).uniqueResult(session);
                }
                catch (PersistenceException e)
                {
                   Throwable cause = e.getCause();
                   if (!(cause instanceof SQLException) || !"42P01".equals(((SQLException) cause).getSQLState()))
                   {
                      throw e; // rethrow
                   }
                }

                if (result != null)
                {
                   nextGlobalID = Math.max((long) result, 1L);
                   System.out.println("Found identity generator sequence starting at " + nextGlobalID + ".");
                }
                else
                {
                   // the sequence generator not found. Remember to create it later.
                   System.out.println("Identity generator sequence not found. Record PKs will start at 1.");
                   createGenerator = true;
                }
                session.commit();

             }
             catch (PersistenceException exc)
             {
                // some error occurred while getting the current value of [p2j_id_generator_sequence]
                System.out.println("Error accessing the id generator sequence. Record PKs will start at 1.");
                createGenerator = true;

                try
                {
                   session.rollback();
                }
                catch (PersistenceException e)
                {
                   LOG.log(Level.SEVERE, "", e);
                }
             }
             finally
             {
                try
                {
                   session.close();
                }
                catch (PersistenceException e)
                {
                   LOG.log(Level.SEVERE, "", e);
                }
             }

             // This iteration is guaranteed to visit bundles which are
             // dependencies of other tables, before visiting their dependent bundles.
             Iterator<ImportBundle> iter = importBundles.values().iterator();
             while (iter.hasNext())
             {
                ImportBundle bundle = iter.next();
                totalBytes += bundle.getFileLength();
                bundle.calculateWeight();
             }

             List<ImportBundle> sorted = new ArrayList<>(importBundles.values());

             // Calculate relative ranks for each import bundle.  This will
             // determine the order in which the tables are loaded.
             iter = sorted.iterator();
             while (iter.hasNext())
             {
                ImportBundle next = iter.next();
                next.calculateRank();
             }

             // sort in priority order, the highest priority first.
             Collections.sort(sorted);

             // report the order in which the bundles will be performed
             System.out.println("IMPORT ORDER:");
             iter = sorted.iterator();
             for (int i = 1; iter.hasNext(); i++)
             {
                ImportBundle next = iter.next();
                System.out.println("\t" + i + ") " + next);
             }

             // Set up the bundle queue.
             bundleQueue = new LinkedList<>(sorted);
             ImportBundle bundle = null;

             Set<String> regularTables = SystemTable.filterTables(SystemTable.REGULAR);

             // schedule asynchronous imports of the import bundles, in order
             while ((bundle = nextBundle()) != null)
             {
                importAsync(bundle, dorder, yearOffset, regularTables);
             }

             // wait for all bundles to finish, reporting any fatal errors which were encountered. Create a
             // sequence for identity generation, using the high-water mark ID created during the import
             synchronized (complete)
             {
                while (!complete.containsAll(attempted))
                {
                   complete.wait();
                }

                createIdentitySequence();
             }

             // return the number of records imported across all bundles
             synchronized (countLock)
             {
                return imported;
             }
          }

          /**
           * Reports any error encountered. If no errors were encountered, the method returns {@code null}
           * immediately.
           *
           * @return  A string containing a formatted summary of the errors encountered for each table.
           *
           */
          public String reportErrors()
          {
             synchronized (countLock)
             {
                if (errors.isEmpty())
                {
                   return "Number of failed records: 0.";
                }

                Set<Map.Entry<String, Integer>> entries = errors.entrySet();
                int totalErrors = 0;
                for (Map.Entry<String, Integer> entry : entries)
                {
                   totalErrors += Math.abs(entry.getValue());
                }

                if (totalErrors == 0)
                {
                   return "Number of failed records: 0."; // this should not happen!
                }

                StringBuilder sb = new StringBuilder();
                sb.append("Number of failed records: ").append(totalErrors).append(". Summary by table:");

                for (Map.Entry<String, Integer> entry : entries)
                {
                   Integer val = entry.getValue();
                   sb.append("\n\t").append(entry.getKey()).append(": ").append(val);
                   if (fatal.contains(entry.getKey()))
                   {
                      sb.append(" (fatal)");
                   }
                }

                return sb.toString();
             }
          }

          /**
           * Adds a sequence to the map of sequences to be initialized from the sequences export file.
           *
           * @param   historical
           *          the original name of the sequence.
           * @param   sqlName
           *          the converted, SQL-compliant name of the sequence.
           */
          public void addSequence(String historical, String sqlName)
          {
             sequences.put(historical, sqlName);
          }

          /**
           * Sets the values of the  sequences. All sequences must be already added to the list.
           *
           * @param   path
           *          Path to the file to which the sequences' values have been exported
           * @param   seqDumpFile
           *          The file to which the sequences' values have been exported
           *
           * @return  The number of successfully initialized
           */
          public long initializeSequences(String path, String seqDumpFile)
          {
             long count = 0;

             Scanner scanner;
             try
             {
                StringBuilder sb = new StringBuilder(path);
                if (!path.endsWith("/"))
                {
                   sb.append("/");
                }
                sb.append(seqDumpFile);
                scanner = new Scanner(new FileInputStream(sb.toString()));
             }
             catch (FileNotFoundException e)
             {
                return 0;
             }

             Session session = null;
             try
             {
                while (scanner.hasNextLine())
                {
                   String line = scanner.nextLine();
                   if (line.equals("."))
                   {
                      break;
                   }

                   String[] tokens = line.split("\\s");
                   if (tokens.length != 3)
                   {
                      // malformed line, must have exactly 3 tokens
                      System.out.printf("%s: The file may not be a sequence export file%n", seqDumpFile);
                      continue;
                   }

                   // interpreting the 3 tokens
                   int seqId = Integer.parseInt(tokens[0]);

                   // remove both " from begin and end
                   String histName = tokens[1].substring(1, tokens[1].length() - 1);
                   long seqVal = Long.parseLong(tokens[2]);

                   if (!sequences.containsKey(histName))
                   {
                      // that's strange, the two lists do not match
                      System.out.printf("%s: Unknown properties for sequence '%s'. The sequence does not " +
                            "appear in .df export so its value will be ignored.%n", seqDumpFile, histName);
                      continue;
                   }

                   String sqlName = sequences.get(histName);
                   String query = dialect.getSequenceSetValString(sqlName, seqVal);
                   boolean inTx = false;

                   try
                   {
                      if (session == null)
                      {
                         // lazy initialization
                         session = openSession();
                      }

                      System.out.printf("%s: Importing sequence %s...%n", seqDumpFile, sqlName);
                      inTx = session.beginTransaction();
                      SQLQuery sqlQ = Session.createSQLQuery(query);
                      if (query.trim().toUpperCase().startsWith("SELECT"))
                      {
                         // a select statement for this likely means we are calling a function to set
                         // the value of the sequence;  we don't care about the result, just the
                         // side-effect of executing the function
                         // TODO: create the proper List<RowStructure> parameter or find other API to call
                         sqlQ.uniqueResult(session);
                      }
                      else
                      {
                         sqlQ.executeUpdate(session);
                      }
                      session.commit();

                      count++; // increment counter on success
                      sequences.remove(histName);
                   }
                   catch (PersistenceException exc)
                   {
                      if (inTx)
                      {
                         try
                         {
                            session.rollback();
                         }
                         catch (PersistenceException rbEx)
                         {
                            LOG.log(Level.FINE,
                                    "Failed rollback operation for create of '" + sqlName + "' sequence.", rbEx);
                         }
                      }

                      LOG.log(Level.INFO, "Error adding sequence '" + sqlName + "'.", exc);
                   }
                }
             }
             finally
             {
                scanner.close();
                if (session != null && session.isOpen())
                {
                   try
                   {
                      session.close();
                   }
                   catch (PersistenceException exc)
                   {
                      LOG.log(Level.FINE, "", exc);
                   }
                }
             }

             // double check if all sequence have been initialized
             if (!sequences.isEmpty())
             {
                System.out.println(seqDumpFile + ": The following sequences were NOT initialized because they " +
                                   "were not found in the export file or import failed:");
                for (Map.Entry<String, String> entry : sequences.entrySet())
                {
                   String seq = entry.getKey();
                   String sqlSeq = entry.getValue();
                   if (!seq.equals(sqlSeq))
                   {
                      // name differ, let the user know the SQL AND legacy name
                      System.out.format("\t%s (%s)%n", sqlSeq, seq);
                   }
                   else
                   {
                      // otherwise, keep it as simple as possible
                      System.out.println("\t" + seq);
                   }
                }
             }

             return count;
          }

          /**
           * Import all data for a table into the database, using the given record loader and current session
           * factory. Uses the {@link com.goldencode.p2j.util.Stream#readField} to scan the input file. Each
           * record's data is stored in a new instance of a data model object (DMO).
           * <p>
           * If the DMO refers to any other DMOs, the necessary queries are
           * executed to resolve these references before the object is
           * persisted. This ensures foreign key associations are correctly
           * resolved at import time.
           * <p>
           * The import uses JDBC batching to send data to the database in batches rather than a single record
           * at a time. Batch size is retrieved from the configuration, and is set to a default if not specified
           * in the config.
           * <p>
           * This method is designed to be fairly fault tolerant. Database errors which are the result of data
           * which does not conform to the expected schema generally are handled by dropping the offending record,
           * adding a log entry to report the nature of the error, and continuing with the next record, if any.
           * Specifically, the following conditions are recoverable:
           * <ul>
           *    <li>character encoding error on read if the specific byte is mapped (see {@code mapBytes});
           *    <li>insert which violates an explicit unique constraint;
           *    <li>insert which violates an explicit not-null constraint;
           *    <li>query which is supposed to return a unique result returns more than one result.
           * </ul>
           * <p>
           * The following errors are not recoverable:
           * <ul>
           *    <li>data which cannot be lexed;
           *    <li>data which does not match the expected lexical token type;
           *    <li>any other exceptions thrown by the database.
           * </ul>
           * In the event an unrecoverable error occurs, the records in the current batch and all the remaining
           * records for the current table are not imported. However, all records imported up to that point
           * are preserved in the database.
           * <p>
           * Errors are logged as configured for the current VM.
           *
           * @param   loader
           *          Record loader helper which contains all information necessary to find the import data and to
           *          map each record in the import file to a DMO of the appropriate class.
           * @param   indexes
           *          A list of index definitions which must be applied to the table before (for unique indexes)
           *          and after (for non-unique indexes) data import.
           * @param   isMandatory
           *          {@code true} if the table must be imported even if we are executing an empty import,
           *          {@code false} otherwise.
           *
           * @return  The number of records successfully imported.
           */
          public int importTable(SqlRecordLoader loader, List<P2JIndex> indexes, boolean isMandatory)
          {
             IDContext idCtx;
             synchronized (idContextStack)
             {
                idCtx = idContextStack.peek() != null ? idContextStack.pop() : new IDContext();
             }
             long nextID = idCtx.getCurrentID();
             Session session = null;
             DataFileReader stream = null;
             List<BaseRecord> records = new ArrayList<>(batchSize);
             Map<BaseRecord, Long> dOffsets = new IdentityHashMap<>(batchSize);
             File dataFile = loader.getDataFile();
             int counter = 0;
             int badIndex = -1;
             int dropCount = 0;

             String fileName = dataFile.getName();
             String table = loader.getTable();
             // Skip the record processing entirely while doing an empty import.
             // This is done per table since table indexes and sequences should still be generated.
             boolean shouldImport = !empty || isMandatory;
             try
             {
                List<WordTableInfo> wordTables = indexes.stream().filter(P2JIndex::isWord).
                      filter(index -> index.components().size() > 0).
                      map(idx -> new WordTableInfo(loader, idx)).
                      filter(wt -> wt.isValid).
                      collect(Collectors.toList());
                // Attempt to drop indexes.
                List<P2JIndex> nonRedundantIndexes = P2JIndex.nonRedundantIndexes(indexes);
                dropIndexes(nonRedundantIndexes, fileName);

                // Attempt to create unique indexes before data import.
                List<P2JIndex> minimalUnique = P2JIndex.normalizeUniqueIndexes(indexes, P2JIndexComponent.COLUMN);
                createIndexes(minimalUnique, true, fileName);
                Map<String, BaseDataType> notNullProperties = loader.notNullProps;
                boolean hasNotNull = notNullProperties.size() != 0;

                if (shouldImport)
                {
                   // process data records
                   stream = new DataFileReader(dataFile.getPath());
                   if (mapBytes != null)
                   {
                      stream.setInvalidBytesMapping(mapBytes);
                   }

                   long expectedRecords = loader.isRecordEmpty() ? 0 : -1;
                   if (stream.pscHeader != null)
                   {
                      String recordsNo = stream.getMetadata("records");
                      if (recordsNo != null)
                      {
                         try
                         {
                            expectedRecords = Long.parseLong(recordsNo);
                         }
                         catch (NumberFormatException e)
                         {
                            System.out.printf("%s: Failed to parse the number of records in PSC footer%n", fileName);
                         }
                      }
                   }

                   boolean recovery = false;
                   boolean eof = false;
                   while (true)
                   {
                      // reset bad index to clear error status;  will be set to a non-negative number on error
                      badIndex = -1;

                      // Open a new session and begin a transaction.
                      session = openSession();
                      session.disableCache();
                      boolean inTx = session.beginTransaction();

                      // Read up to batchSize records from input file.
                      for (int i = records.size(); i < batchSize && !eof && !recovery; i++, counter++)
                      {
                         // the offset at which the record is read from file
                         long crtOffset = stream.getPos();

                         // read the next record
                         Record dmo = null;
                         try
                         {
                            // NOTE: looks like the _User meta-table suffers changes from one version of P4GL to
                            //       another. The good part is that the changes are 'additive': new columns are added
                            //       to newer versions. The way nextRecord is implemented, if the number of column do
                            //       not match the number of read fields, only the first common are initialized, the
                            //       remaining fields od dmo will have the default values for respective data types.
                            dmo = loader.nextRecord(stream);

                            if (dmo != null && expectedRecords >= 0 && counter >= expectedRecords)
                            {
                               if (loader.isRecordEmpty())
                               {
                                  // detected 'reading' [expectedRecords] empty records. Stopping here, drop last one.
                                  eof = true;
                                  break;
                               }

                               // reading more records than expected: log event but process all of them
                               System.out.printf("%s: Read more records than declared in PSC footer (%d)%n",
                                                 fileName, expectedRecords);
                               expectedRecords = -2; // to disable subsequent messages
                            }
                         }
                         /*
                         catch (NonUniqueResultException exc)
                         {
                            log.log(Level.SEVERE,
                                    "Dropped record #" + (counter + 1) + " in " +
                                    dataFile.getName() + " due to error:  " + exc.getMessage());

                            badIndex = counter;
                            dropCount++;

                            // This error is recoverable; no need to discard session, just continue.
                            continue;
                         }
                         */
                         catch (Exception exc)
                         {
                            if (session.isOpen())
                            {
                               session.close();
                            }
                            session = null;

                            // not recoverable: log and abort
                            System.out.printf(
                                  "%s: Error creating record #%d (%d previously dropped) [file offset: %#010x]. Cause: %s%n",
                                  fileName, nextID, dropCount, crtOffset, exc.getMessage());

                            throw exc;
                         }

                         if (dmo == null)
                         {
                            eof = true;
                            break;
                         }

                         if (hasNotNull)
                         {
                            // for all not-null properties which don't have data, set it to default
                            for (Map.Entry<String, BaseDataType> entry : notNullProperties.entrySet())
                            {
                               String prop = entry.getKey();
                               BaseDataType def = entry.getValue();
                               if (def != null)
                               {
                                  loader.overwriteNullProperty(prop, dmo, def);
                               }
                            }
                         }

                         records.add(dmo);
                         dOffsets.put(dmo, crtOffset);
                      }

                      // abort if nothing to do
                      if (records.size() == 0)
                      {
                         break;
                      }

                      // Import the batch as a discrete transaction.
                      int index = 0;
                      BaseRecord dmo = null;
                      Long offset = null;
                      try
                      {
                         Functions.setCodePage(dialect.getCodepage(session.getConnection()));
                         // recovery import (batches of 1) until we've dropped the record which broke
                         // the last batch
                         while (recovery && !records.isEmpty())
                         {
                            badIndex++;

                            if (!inTx)
                            {
                               inTx = session.beginTransaction();
                            }

                            dmo = records.remove(0);
                            offset = dOffsets.remove(dmo);
                            session.save(dmo, false);
                            if (dialect.useWordTables())
                            {
                               populateWordTables(session, dmo, wordTables, fileName);
                            }

                            // commit transaction
                            if (inTx)
                            {
                               session.commit();
                               inTx = false;
                            }
                         }

                         recovery = false;

                         // normal import in batch
                         if (inTx)
                         {
                            // assign a primary key to each record
                            for (int size = records.size(); index < size; index++)
                            {
                               dmo = records.get(index);
                               nextID = idCtx.getNextID();
                               dmo.primaryKey(nextID);
                            }

                            // insert records using SQL batching
                            session.bulkSave(records);

                            if (dialect.useWordTables())
                            {
                               for (BaseRecord dmo2 : records)
                               {
                                  populateWordTables(session, dmo2, wordTables, fileName);
                               }
                            }

                            // commit transaction
                            session.commit();

                            /*
                            // Primitive progress indicator.
                            if ((counter - dropCount) % (increment) == 0)
                            {
                               System.out.print(".");
                            }
                            */

                            // Clear record list for next pass.
                            records.clear();
                            dOffsets.clear();
                         }
                      }
                      catch (UniqueResultException exc)
                      {
                         System.out.printf("%s: Dropped record #%d at file offset %#010x due to error: %s%s%n",
                                           fileName,
                                           counter + 1,
                                           offset == null ? -1 : offset,
                                           exc.getMessage(),
                                           (dmo == null ? "" : "\n\t" + ((Record) dmo).toString2(true)));

                         badIndex = counter;
                         dropCount++;

                         // This error is recoverable; no need to discard session, just continue.
                         continue;
                      }
                      catch (PersistenceException exc)
                      {
                         session.rollback();

                         if (recovery)
                         {
                            dropCount++;

                            // log this as a severe import event and the error message
                            System.out.printf("%s: Dropped record #%d at file offset %#010x due to error: %s%s%n",
                                              fileName,
                                              counter - Math.min(counter, batchSize) + dropCount + badIndex,
                                              offset == null ? -1 : offset,
                                              exc.getMessage(),
                                              (dmo == null ? "" : "\n\t" + ((Record) dmo).toString2(true)));
                         }
                         else
                         {
                            // we hit a failing record importing a batch; go into recovery mode
                            recovery = true;
                         }
                      }
                      finally
                      {
                         Functions.setCodePage(null);
                         if (session.isOpen())
                         {
                            session.close();
                         }
                         session = null;
                      }
                   }
                }

                // attempt to create remaining indexes
                Set<P2JIndex> remainingIndexes = new LinkedHashSet<>(nonRedundantIndexes);
                remainingIndexes.removeAll(minimalUnique);
                createIndexes(remainingIndexes, false, fileName);

                // if supported by dialect, analyze the table to speed queries which leverage indexes
                // If we are running an empty import, there will be no records and analyzing the table
                // will not be necessary.
                String analyzeStmt = dialect.getAnalyzeString(table);
                if (analyzeStmt != null && shouldImport)
                {
                   Connection connection = session.getConnection();
                   try (Statement statement = connection.createStatement())
                   {
                      System.out.printf("%s: Analyzing %s...%n", fileName, table);
                      SQLExecutor.getInstance().execute(session.getDatabase(),
                                                        Statement::execute,
                                                        statement,
                                                        analyzeStmt);
                      connection.commit();
                   }
                   catch (SQLException exc)
                   {
                      connection.rollback();
                      Throwable next = exc.getNextException();
                      if (next != null)
                      {
                         System.out.printf(
                               "%s: Failed to analyze table '%s': %s%n", fileName, table, next.getMessage());
                      }

                      throw exc;
                   }
                }
             }
             catch (Exception exc)
             {
                int processed = adjustedCounter(counter, badIndex, dropCount);
                int lost = counter % batchSize;
                long recCount;
                if (stream == null && shouldImport)
                {
                   recCount = -1;
                   dropCount++; // at least one
                }
                else
                {
                   recCount = stream.getRecordCount();
                   dropCount += (recCount - processed);
                }
                System.out.printf(
                      "%s: Error processing import data; %d of %s record(s) successfully processed;  " +
                      "%s record(s) uncommitted due to this error;  %d record(s) dropped. Cause: %s%n",
                      fileName, processed, (recCount < 0 ? "?" : recCount), lost, dropCount, exc.getMessage());
                LOG.log(Level.INFO, "", exc);

                synchronized (complete)
                {
                   fatal.add(loader.getDataFile().getName());
                }
             }
             finally
             {
                if (session != null && session.isOpen())
                {
                   try
                   {
                      session.close();
                   }
                   catch (PersistenceException exc)
                   {
                      LOG.log(Level.FINE, "", exc);
                   }
                }
                if (stream != null)
                {
                   stream.close();
                }
                synchronized (idContextStack)
                {
                   idContextStack.push(idCtx);
                }
             }

             if (dropCount != 0)
             {
                synchronized (countLock)
                {
                   errors.put(fileName, dropCount);
                }
             }

             // Report number of records successfully imported.
             return adjustedCounter(counter, badIndex, dropCount);
          }

          /**
           * Populate word tables with data from the record.
           *
           * @param   session
           *          Database session.
           * @param   dmo
           *          Master record.
           * @param   wordTables
           *          List containing data to be used when populating the word tables.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  PersistenceException
           *          if a FWD internal problem occurred.
           * @throws  SQLException
           *          if a SQL related problem occurred.
           */
          private void populateWordTables(Session session,
                                          BaseRecord dmo,
                                          List<WordTableInfo> wordTables,
                                          String fileName)
          throws PersistenceException, SQLException
          {
             if (wordTables.isEmpty())
             {
                return;
             }
             Object[] data = dmo.getData(wordTables.get(0).mapper);
             if (data == null)
             {
                return;
             }
             Connection conn = session.getConnection();
             for (WordTableInfo wordTable: wordTables)
             {
                if (wordTable.extent == 0)
                {
                   String value = (String) data[wordTable.offset];
                   // NB: uppercase at the database side
                   String[] words = Functions.words(value, false, !wordTable.caseSensitive);
                   populateWordTable(conn, wordTable.wordTableName, dmo.primaryKey(),
                         words, !wordTable.caseSensitive, fileName);
                }
                else
                {
                   for (int i = 0; i < wordTable.extent; i++)
                   {
                      String value = (String) data[wordTable.offset + i];
                      // NB: uppercase at the database side
                      String[] words = Functions.words(value, false, !wordTable.caseSensitive);
                      populateWordTable(conn, wordTable.wordTableName, dmo.primaryKey(), i + 1,
                            words, !wordTable.caseSensitive, fileName);
                   }
                }
             }
          }

          /**
           * Populate word tables with data.
           *
           * @param   conn
           *          Database connection.
           * @param   wordTableName
           *          The word table name.
           * @param   primaryKey
           *          The primary key of the master record.
           * @param   words
           *          The words in the field.
           * @param   toUppercase
           *          Instruct to convert words to uppercase.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  SQLException
           *          if a SQL related problem occurred.
           */
          private void populateWordTable(Connection conn,
                                         String wordTableName,
                                         Long primaryKey,
                                         String[] words,
                                         boolean toUppercase,
                                         String fileName)
          throws SQLException
          {
             if (words == null || words.length == 0)
             {
                return;
             }

             StringBuilder sql = new StringBuilder("INSERT INTO ").append(wordTableName).append(" VALUES");
             for (int i = 0; i < words.length; i++)
             {
                // NB: uppercase at the database side
                sql.append(i == 0 ? "" : ",").append(toUppercase ? "(?,UPPER(?))" : "(?,?)");
             }

             Savepoint sp = conn.setSavepoint();
             try (PreparedStatement pstmt = conn.prepareStatement(sql.toString()))
             {
                int i = 1;
                for(String word: words)
                {
                   pstmt.setLong(i++, primaryKey);
                   pstmt.setString(i++, word);
                }
                pstmt.executeUpdate();
             }
             catch (SQLException e)
             {
                LOG.log(Level.INFO,
                        String.format("%s: Failed to populate word table %s: pk = %d.%n",
                                      fileName,
                                      wordTableName,
                                      primaryKey),
                        e);
                conn.rollback(sp);
             }
          }

          /**
           * Populate word tables with data.
           *
           * @param   conn
           *          Database connection.
           * @param   wordTableName
           *          The word table name.
           * @param   primaryKey
           *          The primary key of the master record.
           * @param   index
           *          The index in the extent.
           * @param   words
           *          The words in the field.
           * @param   toUppercase
           *          Instructs to convert words to uppercase.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  SQLException
           *          if a SQL related problem occurred.
           */
          private void populateWordTable(Connection conn,
                                         String wordTableName,
                                         Long primaryKey,
                                         int index,
                                         String[] words,
                                         boolean toUppercase,
                                         String fileName)
          throws SQLException
          {
             if (words == null || words.length == 0)
             {
                return;
             }

             StringBuilder sql = new StringBuilder("INSERT INTO ").append(wordTableName).append(" VALUES");
             for (int i = 0; i < words.length; i++)
             {
                // NB: uppercase at the database side
                sql.append(i == 0 ? "" : ",").append(toUppercase ? "(?,?,UPPER(?))" : "(?,?,?)");
             }

             Savepoint sp = conn.setSavepoint();
             try(PreparedStatement pstmt = conn.prepareStatement(sql.toString()))
             {
                int i = 1;
                for(String word: words)
                {
                   pstmt.setLong(i++, primaryKey);
                   pstmt.setInt(i++, index);
                   pstmt.setString(i++, word);
                }
                pstmt.executeUpdate();
             }
             catch(SQLException e)
             {
                System.out.printf("%s: Failed to populate word table %s: pk = %d, index = %d.%n",
                                  fileName, wordTableName, primaryKey, index);
                LOG.log(Level.FINE, "Failed to populate word table " + wordTableName, e);
                conn.rollback(sp);
             }
          }

          /**
           * Commence a word reindex run, potentially on multiple threads. All tables
           * prepared by {@link #prepareImport} will be processed.
           * <p>
           * Multiple, parallel threads are used to process multiple tables at once.
           * <p>
           * This method will block until all prepared tables have been processed.
           *
           * @throws  InterruptedException
           *          if the thread on which this method is invoked is interrupted.
           *
           * @see     #prepareImport
           * @see     #setMaximumThreads
           */
          public void runWordReindex()
          throws InterruptedException
          {
             if (!dialect.useWordTables())
             {
                LOG.log(Level.FINE, "Dialect for " + dialect.getDriverClassName() +
                         " does not use word index tables.");
                return;
             }

             // Set up the bundle queue.
             bundleQueue = new LinkedList<>(importBundles.values());
             ImportBundle bundle = null;

             // schedule asynchronous word reindex of the import bundles
             while ((bundle = nextBundle()) != null)
             {
                wordReindexAsync(bundle.getLoader(), bundle.getIndexes());
             }

             // wait for all bundles to finish, reporting any fatal errors which were encountered.
             synchronized (complete)
             {
                while (!complete.containsAll(attempted))
                {
                   complete.wait();
                }
             }
          }

          /**
           * Repopulate the word index tables using rebuild_word_index() SQL UDF
           *
           * @param   loader
           *          Record loader helper which contains all information
           *          necessary to find the import data and to map each record
           *          in the import file to a DMO of the appropriate class.
           * @param   indexes
           *          A list of index definitions.
           */
          public void wordReindexAsync(SqlRecordLoader loader, List<P2JIndex> indexes)
          {
             if (!dialect.useWordTables())
             {
                LOG.log(Level.FINE, "Dialect for " + dialect.getDriverClassName() +
                         " does not use word index tables.");
                return;
             }

             Runnable runner = () ->
             {
                try
                {
                   for (P2JIndex index : indexes)
                   {
                      if (!index.isWord())
                      {
                         continue;
                      }
                      WordTableInfo wordTable = new WordTableInfo(loader, index);
                      if (!wordTable.isValid)
                      {
                         continue;
                      }
                      try (Session session = openSession())
                      {
                         LOG.log(Level.INFO, String.format("BEGIN: repopulating word index table <%s> ",
                                 wordTable.wordTableName));
                         session.disableCache();
                         String fieldName = wordTable.originalName.isEmpty() ?
                                  wordTable.fieldName : wordTable.originalName;
                         String sql = (wordTable.extent == 0) ? "SELECT udf.rebuild_word_index(?, ?, ?, ?)" :
                                                                "SELECT udf.rebuild_word_index(?, ?, ?, ?, ?)";
                         Connection conn = session.getConnection();
                         try (PreparedStatement pstmt = conn.prepareStatement(sql))
                         {
                            pstmt.setString(1, wordTable.wordTableName);
                            pstmt.setString(2, index.getTable());
                            pstmt.setString(3, fieldName);
                            pstmt.setBoolean(4, !wordTable.caseSensitive);
                            if (wordTable.extent > 0)
                            {
                               pstmt.setInt(5, wordTable.extent);
                            }
                            pstmt.execute();
                         }
                         LOG.log(Level.INFO, String.format("END: repopulated word index table <%s> ",
                                 wordTable.wordTableName));
                      }
                      catch (PersistenceException | SQLException e)
                      {
                         LOG.log(Level.SEVERE, String.format("Failed to repopulate word index table: <%s> ",
                                 wordTable.wordTableName), e);
                      }
                   }
                }
                catch (Exception e)
                {
                   LOG.log(Level.SEVERE, String.format("Failed to repopulate word index tables for: <%s> ",
                           loader.getTable()), e);
                }

                synchronized (complete)
                {
                   // update the set of completed bundles, regardless of success
                   complete.add(loader.getDMOName());

                   // decrement the number of active threads and notify any waiting threads
                   activeThreads--;
                   complete.notifyAll();
                }
             };

             synchronized (complete)
             {
                // Update the set of attempted import bundles.
                attempted.add(loader.getDMOName());

                // Increment the number of active threads.
                activeThreads++;
             }

             // Start the import bundle thread.
             Thread t = new Thread(runner);
             t.setDaemon(true);
             t.start();
          }

          /**
           * Drop and recreate the list of indexes for the given table.  If the table does not exist in the
           * database, simply return.  Indexes on {@code table} which are not within the list are ignored.
           * <p>
           * Indexes to be dropped are identified by the names provided in the list of index definitions; the
           * remainder of the definition is not used to identify indexes to be dropped.
           *
           * @param   table
           *          Table which is to be reindexed.
           * @param   indexes
           *          List of index definitions.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  PersistenceException
           *          if any error occurs opening or closing sessions, or obtaining database connections.
           */
          public void reindex(String table, List<P2JIndex> indexes, String fileName)
          throws PersistenceException
          {
             List<P2JIndex> minimalUnique = P2JIndex.normalizeUniqueIndexes(indexes, P2JIndexComponent.COLUMN);
             List<P2JIndex> nonRedundant = P2JIndex.nonRedundantIndexes(indexes);
             Set<P2JIndex> remaining = new LinkedHashSet<>(nonRedundant);
             remaining.removeAll(minimalUnique);

             dropIndexes(nonRedundant, fileName);
             createIndexes(minimalUnique, true, fileName);
             createIndexes(remaining, false, fileName);
          }

          /**
           * Get the implementation class mapped to the specified interface.
           *
           * @param   ifaceName
           *          The full qualified interface name for which an implementing class name is being
           *          looked up.
           *
           * @return  DMO class which implements interface {@code iface}, or {@code null} if no such
           *          mapping is found.
           *
           * @throws  ClassNotFoundException
           *          if {@code ifaceName} parameter is not recognized.
           * @throws  IllegalArgumentException
           *          if {@code ifaceName} parameter is not recognized.
           */
          public Class<? extends Record> getDmoClass(String ifaceName)
          throws ClassNotFoundException
          {
             Class<? extends DataModelObject> iface = (Class<? extends DataModelObject>) Class.forName(ifaceName);
             return DmoMetadataManager.registerDmo(iface).getImplementationClass();
          }

          /**
           * Get word table name for the DMO and index name
           * @param ifaceName
           *        DMP interface name
           * @param indexName
           *        index name
           * @return word table name
           */
          public String getWordTableName(String ifaceName, String indexName)
          {
             DmoMeta meta = DmoMetadataManager.getDmoInfo(ifaceName);
             return meta.getWordTablesByIndexName().get(indexName);
          }

          /**
           * Return the next import bundle which is eligible for import.  A bundle is eligible for
           * import if all of the tables upon which it is dependent already have been loaded.  Bundles
           * are returned from this method in the order in which they were ranked, except that a
           * bundle with a lower rank may be returned if a higher ranking bundle still has unresolved
           * dependencies.  If there are no more bundles in the bundle queue, the import run is
           * finished, and this method will return {@code null}.
           * <p>
           * This method will block if the maximum number of active threads has
           * been reached.  It will unblock once a slot becomes available.
           *
           * @return  The next import bundle to be processed, or {@code null} if no bundles remain in
           *          the queue.
           *
           * @throws  InterruptedException
           *          if the running thread is interrupted while waiting for an import thread slot to
           *          become available.
           */
          private ImportBundle nextBundle()
          throws InterruptedException
          {
             synchronized (complete)
             {
                // wait for an available thread slot
                while (activeThreads == maximumThreads)
                {
                   complete.wait();
                }

                if (bundleQueue.isEmpty())
                {
                   // queue is empty. All done. Good job!
                   return null;
                }
                // take the first bundle. Remove it from the queue and return it
                Iterator<ImportBundle> iter = bundleQueue.iterator();
                ImportBundle bundle = iter.next();
                iter.remove();

                return bundle;
             }
          }

          /**
           * Import a single table on a dedicated import thread.  This method will block until a
           * thread slot becomes available, if the maximum number of simultaneous threads already
           * is running.
           * <p>
           * Once a thread slot opens up, a new thread is launched to import the table associated
           * with <code>bundle</code>.  If this table is dependent upon the import of other tables,
           * and not all of these tables have yet been imported, the new thread will block until such
           * time as all such dependencies have been resolved.  If any tables upon which the target
           * table is dependent have encountered fatal errors, the import of the target table is
           * aborted and the target table is added to the set of tables with fatal errors.
           *
           * @param   bundle
           *          Import bundle which holds all information necessary to import data into a
           *          single table.
           * @param   dateOrder
           *          The date format to be used when importing if PSC footer not found in file.
           * @param   yearOffset
           *          The offset year to be used when importing if PSC footer not found in file.
           * @param   regularTables
           *          The set containing all tables that must be imported, even when running an empty
           *          import.
           *
           * @see     #setMaximumThreads
           */
          private void importAsync(final ImportBundle bundle,
                                   final String dateOrder,
                                   final Long yearOffset,
                                   final Set<String> regularTables)
          {
             Runnable runner = () ->
             {
                // setup date parsing specifications, if PSC footer is found in file, they will
                // be over-written when opening the input file
                if (dateOrder != null)
                {
                   SessionUtils.setDateFormat(dateOrder);
                }
                if (yearOffset != null)
                {
                   date.setWindowingYear(yearOffset);
                }

                SqlRecordLoader loader = bundle.getLoader();
                String table = loader.getTable();
                boolean isMandatory = regularTables.contains(loader.getLegacyName().toLowerCase());
                String dmoName = loader.getDMOName();
                Thread.currentThread().setName("Importer for " + table + " table");
                String fileName = loader.getDataFile().getName();
                System.out.printf("BEGIN %s (%s)%n", fileName, table);

                // run the import bundle only if the source file is not empty
                long bytes = bundle.getFileLength();
                int count = (bytes == 0) ? 0 : importTable(loader, bundle.getIndexes(), isMandatory);

                synchronized (complete)
                {
                   synchronized (countLock)
                   {
                      imported += count;
                      importedBytes += bytes;

                      messageWithStats(String.format("END %s. %d records imported into %s", fileName, count, table));
                   }

                   // update the set of completed bundles, regardless of success
                   complete.add(dmoName);

                   // decrement the number of active threads and notify any waiting threads
                   activeThreads--;
                   complete.notifyAll();
                }
             };

             synchronized (complete)
             {
                // Update the set of attempted import bundles.
                attempted.add(bundle.getLoader().getDMOName());

                // Increment the number of active threads.
                activeThreads++;
             }

             // Start the import bundle thread.
             Thread t = new Thread(runner);
             t.setDaemon(true);
             t.start();
          }

          /**
           * Print a message to stdout, followed by the current import statistics.
           * Statistics consist of the number of records imported since the start
           * of the import, and average counts of records imported per second,
           * per minute, and per hour.  Statistics are printed after the core
           * message, on a separate line.
           *
           * @param   message
           *          The core message to be printed.  May be {@code null} if only statistics are
           *          desired.
           */
          private void messageWithStats(String message)
          {
             if (message != null)
             {
                System.out.println(message);
             }

             synchronized (countLock)
             {
                long elapsed = System.currentTimeMillis() - startTime;
                if (elapsed == 0)
                {
                   System.out.println("0 recs in 0:00:00.000");
                   return;
                }

                System.out.printf(STATS_FMT,
                                  totalBytes == 0 ? 100 : (importedBytes / totalBytes * 100),
                                  imported,
                                  elapsed / 3600000,
                                  elapsed / 60000 % 60,
                                  elapsed / 1000.0 % 60,
                                  (long) (imported / (elapsed / 1000.0)),
                                  (long) (imported / (elapsed / 60000.0)),
                                  (long) (imported / (elapsed / 3600000.0)));
             }
          }

          /**
           * Sequentially create the DDL to drop all of the specified indexes from
           * the database, write this SQL to the given stream.
           *
           * @param    os
           *           Output destination.
           * @param    delimiter
           *           End of statement delimiter.
           * @param    indexes
           *           Collection of indexes to be dropped.
           */
          private void generateDropIndexes(OutputStream         os,
                                           String               delimiter,
                                           Collection<P2JIndex> indexes)
          throws IOException
          {
             for (P2JIndex index : indexes)
             {
                writeStmt(os, dialect.getDropIndexString(index), delimiter);
             }
          }

          /**
           * Sequentially drop all of the specified indexes from the database. JDBC errors encountered executing
           * the drop are logged but are otherwise ignored, since it may very well be that the indexes to be
           * dropped have never been created (as is the case with a new import).
           *
           * @param   indexes
           *          Collection of indexes to be dropped.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  PersistenceException
           *          if any error occurs opening or closing a session, or obtaining a database connection.
           */
          private void dropIndexes(Collection<P2JIndex> indexes, String fileName)
          throws PersistenceException
          {
             Iterator<P2JIndex> iter = indexes.iterator();
             Session session = null;
             try
             {
                while (iter.hasNext())
                {
                   P2JIndex index = iter.next();

                   // do not attempt to drop a word index here;  it was never created
                   if (index.isWord())
                   {
                      System.out.printf("%s: Skipping dropping of word index: %s%n", fileName, index.getName());
                      continue;
                   }

                   if (session == null)
                   {
                      session = openSession();
                   }

                   boolean inTx = false;

                   try
                   {
                      inTx = session.beginTransaction();
                      dropIndex(session, index, fileName);
                      if (inTx)
                      {
                         session.commit();
                      }
                   }
                   catch (PersistenceException exc)
                   {
                      if (inTx)
                      {
                         session.rollback();
                      }
                      session.close();
                      session = null;

                      LOG.log(Level.FINE, "", exc);
                   }
                }
             }
             finally
             {
                if (session != null)
                {
                   session.close();
                }
             }
          }

          /**
           * Drop an index based upon the specified definition. A native SQL statement is executed against the
           * database to drop the index. The statement is generated by using the database-specific dialect
           * configured for the process.
           *
           * @param   session
           *          The session to use.
           * @param   index
           *          Definition of the index to be dropped.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  PersistenceException
           *          if an error occurs at the database during execution of the index drop statement.
           */
          private void dropIndex(Session session, P2JIndex index, String fileName)
          throws PersistenceException
          {
             String dropString = dialect.getDropIndexString(index);

             System.out.printf("%s: SQL: %s%n", fileName, dropString);

             Session.createSQLQuery(dropString).executeUpdate(session);
          }

          /**
           * Sequentially create the DDL to create all of the specified indexes from
           * the database, write this SQL to the given stream.
           *
           * @param    os
           *           Output destination.
           * @param    delimiter
           *           End of statement delimiter.
           * @param    indexes
           *           Collection of indexes to be created.
           */
          private void generateCreateIndexes(OutputStream         os,
                                             String               delimiter,
                                             Collection<P2JIndex> indexes,
                                             boolean              unique)
          throws IOException
          {
             for (P2JIndex index : indexes)
             {
                writeStmt(os, dialect.getCreateIndexString(index, unique, true), delimiter);
             }
          }

          /**
           * Create all indexes in the specified list which match the uniqueness specified by
           * {@code unique}. If none match, no indexes are created.
           *
           * @param   indexes
           *          A list of index definitions.
           * @param   unique
           *          If {@code true}, all unique indexes in the list are created; if {@code false}, all
           *          non-unique indexes in the list are created.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           */
          private void createIndexes(Collection<P2JIndex> indexes, boolean unique, String fileName)
          throws PersistenceException
          {
             for (P2JIndex index : indexes)
             {
                // do not create a word index here
                // TODO:  word indexes must be handled separately
                if (index.isWord())
                {
                   System.out.printf("%s: Skipping creation of word index: %s%n", fileName,  index.getName());
                   continue;
                }

                Session session = openSession();
                boolean inTx = false;

                try
                {
                   inTx = session.beginTransaction();
                   createIndex(session, index, unique, fileName);
                   session.commit();
                }
                catch (PersistenceException exc)
                {
                   if (inTx)
                   {
                      session.rollback();
                   }
                   LOG.log(Level.FINE, "", exc);

                   throw exc;
                }
                finally
                {
                   session.close();
                }
             }
          }

          /**
           * Create an index based upon the specified definition. A native SQL statement is executed against the
           * database to create the index. The statement is generated by using database-specific dialect
           * configured for the import.
           *
           * @param   session
           *          The session to use.
           * @param   index
           *          Definition of the index to be created.
           * @param   unique
           *          If {@code true}, create the index as a unique index, {@code false} to omit the unique
           *          constraint.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @throws  PersistenceException
           *          if an error occurs at the database during index creation.
           */
          private void createIndex(Session session, P2JIndex index, boolean unique, String fileName)
          throws PersistenceException
          {
             // TODO:  check if index already exists before executing create statement. We can use
             //        DatabaseMetadata for this, and it would avoid hassles in the event of a restarted import.
             String createString = dialect.getCreateIndexString(index, unique, false);

             System.out.printf("%s: SQL: %s%n", fileName, createString);

             Session.createSQLQuery(createString).executeUpdate(session);
          }

          /**
           * Adjust the given counter to take the specified <code>badIndex</code>
           * and <code>dropCount</code> values into account.
           *
           * @param   counter
           *          Number of total records read, but not necessarily imported.
           * @param   badIndex
           *          Index of the first problem record in an import batch, or
           *          -1 if there was no problem record. The index is zero-based,
           *          relative to the beginning of the batch.
           * @param   dropCount
           *          Cumulative number of problem records which have been
           *          dropped from the import of the current table.
           *
           * @return  Number of records successfully imported for the current table.
           */
          private int adjustedCounter(int counter, int badIndex, int dropCount)
          {
             if (badIndex >= 0)
             {
                return (counter - batchSize + badIndex);
             }

             return (counter == 0) ? 0 : counter - dropCount;
          }

          /**
           * Remove a problem record from a batch of records which failed on import, if it can be
           * determined which record caused the problem. This is determined by walking the root cause
           * exceptions chained to the exception {@code exc} until an exception of type
           * {@link BatchUpdateException} is found. If such a cause is found, the number of
           * successful updates is extracted. This number represents the index of the first record
           * which was not successful. It is this record which is removed from the batch.
           * <p>
           * An error message is logged recording the 1-based index of the problem record, the data
           * file from which it was read, and the nature of the error as extracted from {@code exc}.
           * <p>
           * The purpose of this method is to preserve information regarding the failure, while
           * dropping the problem record in order to clean up the current batch of records to enable
           * a retry of the batch insert.
           *
           * @param   records
           *          List of records which comprised the failed batch.
           * @param   exc
           *          Exception which was caught on batch insert failure.
           * @param   counter
           *          Number of total records read up to this point. This number will be at most
           *          {@code batchSize} higher than the number of records successfully imported to
           *          this point.
           * @param   fileName
           *          The name of the file where the data is read. Used for debugging.
           *
           * @return  The zero-based index, relative to the beginning of {@code records}, of the
           *          problem record.
           *
           * @throws  PersistenceException
           *          in the event {@code exc} doesn't contain a chained {@link BatchUpdateException}.
           *          In this case, the index of a problem record cannot be determined, and it is
           *          likely that the error is more severe than a single bad record.
           *
           * @deprecated Somewhere along the way, the mechanism we were using to determine which
           *             statement in the batch failed stopped working
           *             ({@code BatchUpdateException.getUpdateCounts} now reports
           *             {@code Statement.EXECUTE_FAILED} for every statement in the batch). So, this
           *             method is no longer reliable.
           */
          @Deprecated
          private int removeFailingRecord(List<Object> records,
                                          PersistenceException exc,
                                          int counter,
                                          String fileName)
          throws PersistenceException
          {
             // try to isolate and remove offending record.
             BatchUpdateException bue = null;
             Throwable next = exc;
             while (next != null)
             {
                if (next instanceof BatchUpdateException)
                {
                   bue = (BatchUpdateException) next;
                   break;
                }
                next = next.getCause();
             }

             // unexpected problem; bail
             if (bue == null)
             {
                throw exc;
             }

             // if a query of the batch failed, its update count will be <> 1 (most likely negative)
             int[] updateCounts = bue.getUpdateCounts();
             int badIndex = 0;
             while (badIndex < updateCounts.length && updateCounts[badIndex] > 0)
             {
                // find the update query that failed to store (a) record into table
                badIndex++;
             }

             records.remove(badIndex); // remove offending record
             Throwable nextErr = bue.getNextException();
             if (nextErr == null)
             {
                nextErr = bue;
             }

             // log this as a severe import event and the error message
             int recNo = counter - Math.min(counter, batchSize) + badIndex + 1;
             System.out.printf(
                   "%s: Dropped record #%d due to error: %s%n", fileName, recNo, nextErr.getMessage());

             return badIndex;
          }
       }

       /**
        * Helper class which manages all aspects of the mapping between a
        * Progress data export file and the DMO which corresponds with a
        * particular database table. The information managed by this class
        * includes:
        * <ul>
        * <li>the DMO class to be instantiated for each imported record;
        * <li>the input data file from which records are read;
        * <li>whether the export file reading process should assert the type
        *     of each data component it reads for a record (for debugging);
        * <li>a list of <code>PropertyMapper</code> objects, in the order the
        *     mapped data is expected to be encountered in the export file for
        *     each record (one per column, or one per subscript value for extent
        *     columns);
        * <li>a map of <code>QueryHelper</code> objects used to resolve foreign
        *     key associations, indexed by the name of the "foreign" class;
        * <li>a map of <code>Method</code> objects used to set a "foreign" class
        *     as property of the data model object currently being imported.
        * </ul>
        * <p>
        * An instance of this class is created for each table which needs to be
        * imported. The above properties are set as necessary, using the various
        * set, put, and add methods. Once the instance is fully configured,
        * {@link #nextRecord} is called repeatedly, until there are no more
        * records to process in the input data file.
        */
       public final static class SqlRecordLoader
       extends RecordLoader
       {
          /** Name of database table */
          private final String table;

          /** Legacy name of database table */
          private final String legacyName;

          /** Map of not-null properties to their default values. */
          private final Map<String, BaseDataType> notNullProps = new HashMap<>();

          /**
           * Constructor.
           * <p>
           * Note: the {@code checkTypes} flag should be set to {@code false} for normal operations. It is
           *       expensive and should be enabled for debug purposes only.
           *
           * @param   dmoClass
           *          Class which is instantiated for each record imported.
           * @param   table
           *          Name of the table associated with this loader instance.
           * @param   legacyName
           *          Legacy name of the table associated with this loader instance.
           * @param   dataFile
           *          Data input file.
           * @param   checkTypes
           *          {@code true} to assert that the data type of each field read matches the expected type.
           */
          public SqlRecordLoader(Class<? extends Record> dmoClass,
                                 String table,
                                 String legacyName,
                                 File dataFile,
                                 boolean checkTypes)
          {
             super(dmoClass, dataFile, checkTypes);

             this.table = table;
             this.legacyName = legacyName;
          }

          /**
           * Add the not-null property with its default value, if any.
           *
           * @param   prop
           *          The property name.
           * @param   def
           *          The default value.
           */
          public void addNotNullProperty(String prop, BaseDataType def)
          {
             notNullProps.put(prop, def);
          }

          /**
           * Get the name of the database table associated with this loader instance.
           *
           * @return  Table name.
           */
          public String getTable()
          {
             return table;
          }

          /**
           * Get the legacy name of the database table associate with this loader instance.
           *
           * @return  Legacy name.
           */
          public String getLegacyName()
          {
             return legacyName;
          }

          /**
           * Process the data value most recently read from the data export file.
           * This method is called by the superclass' <code>nextRecord</code>
           * method for each data value read from the data export file. If the
           * current property is not delegated and not dropped,
           *
           * @param   mapper
           *          Mapping object associated with the current property.
           * @param   record
           *          Java-bean-like object which holds the contents of the record
           *          currently being loaded.
           * @param   value
           *          Data value read from the data export file for the current property. It has
           *          <i>not</i> yet been stored in the {@code record} object.
           */
          @Override
          protected void processNextValue(PropertyMapper mapper, Record record, BaseDataType value)
          {
             // set the value of the data in the DMO
             mapper.setValue(record, value);
          }
       }

       /**
        * All information necessary to perform the import of a single table, and to prioritize that
        * import among all tables to be imported in the same run.
        */
       private static class ImportBundle
       implements Comparable<ImportBundle>
       {
          /** Record loader object */
          private final SqlRecordLoader loader;

          /** List of all indexes to be created for the target table */
          private final List<P2JIndex> indexes;

          /** Length in bytes of the exported data file */
          private final long fileLength;

          /** Load time heuristic */
          private double weight = 0.0;

          /** Import order sorting heuristic */
          private double rank = -1.0;

          /**
           * Constructor.
           *
           * @param   loader
           *          Record loader object.
           * @param   indexes
           *          List of all indexes to be created for the target table.
           */
          ImportBundle(SqlRecordLoader loader, List<P2JIndex> indexes)
          {
             this.loader = loader;
             this.indexes = indexes;
             File file = loader.getDataFile();
             this.fileLength = (file != null && file.exists()) ? file.length() : 0L;
          }

          /**
           * Compare this object with another of the same type for sorting
           * purposes.  Import bundles with the highest calculated rank heuristic
           * are sorted first, so that they will be scheduled for import earlier
           * in the import bundle.
           *
           * @param   that
           *          <code>ImportBundle</code> with which to compare this
           *          instance for sorting purposes.
           *
           * @return  Positive number to sort this instance before the given
           *          object;  negative number to sort this instance after the
           *          given object;  0 to give both objects equal weight.
           */
          @Override
          public int compareTo(ImportBundle that)
          {
             String thisName = this.loader.getDMOName();
             String thatName = that.loader.getDMOName();

             if (this.rank > that.rank)
             {
                return -1;
             }

             if (that.rank > this.rank)
             {
                return 1;
             }

             return thisName.compareTo(thatName);
          }

          /**
           * Produce a string representation of this object, primarily for debug
           * purposes.
           *
           * @return  String which describes this object.
           */
          @Override
          public String toString()
          {
             return loader.getDataFile().getName() + ":" + rank + " [" + fileLength + " bytes]";
          }

          /**
           * Get the record loader associated with this bundle.
           *
           * @return  Record loader.
           */
          SqlRecordLoader getLoader()
          {
             return loader;
          }

          /**
           * Get the indexes to be created for the target table.
           *
           * @return  Indexes.
           */
          List<P2JIndex> getIndexes()
          {
             return indexes;
          }

          /**
           * Get this import bundle's weight, which is a heuristic intended to
           * measure the relative time it will take to load this table.
           *
           * @return  Weight heuristic.
           */
          double getWeight()
          {
             return weight;
          }

          /**
           * Get this import bundle's rank, which is a heuristic intended to rank
           * the importance of loading this table sooner than other tables.
           *
           * @return  Rank heuristic.
           */
          double getRank()
          {
             return rank;
          }

          /**
           * Get the number of bytes in the exported data file for this table.
           *
           * @return  Byte count.
           */
          long getFileLength()
          {
             return fileLength;
          }

          /**
           * Calculate this import bundle's weight, which is a heuristic intended
           * to measure the relative time it will take to load this table.  In
           * making this determination, the weight of each of this table's
           * dependencies is factored in, each of which may have the weight of
           * its dependencies factored in, and so on.  A bundle's weight is used
           * downstream in ranking bundles relative to one another when
           * determining import order.
           *
           * @see  #calculateRank
           */
          void calculateWeight()
          {
             weight = fileLength * 2;
          }

          /**
           * Calculate this import bundle's rank, which is a heuristic intended
           * to rank the importance of loading this table sooner than other
           * tables.  In making this determination, this bundle's weight is taken
           * into consideration (calculated previously), as are the ranks of all
           * those tables which are dependent upon this table to load first.
           * This algorithm ensures that a table always has a higher rank (and
           * thus is sorted to load before) any tables which are dependent upon
           * it.
           *
           * @see  #calculateWeight
           */
          void calculateRank()
          {
             if (rank < 0.0)
             {
                rank = weight;
             }
          }
       }

       /**
        * Thread local context which tracks next available primary key ID.  IDs
        * are unique across all import threads and are batched together within
        * increments of {@link ImportWorker#idBracketSize}.
        */
       private class IDContext
       {
          /** Counter */
          private long counter;

          /**
           * Default constructor;  sets up internal counter.
           */
          IDContext()
          {
             prepareNextBracket();
          }

          /**
           * Get the next available, globally unique ID.  When next ID falls on
           * a bracket boundary, it is returned, but the next bracket of IDs for
           * this thread is prepared.
           *
           * @return  Next, globally unique ID.
           */
          long getNextID()
          {
             long result = ++counter;

             if ((result % idBracketSize) == 0)
             {
                prepareNextBracket();
             }

             return result;
          }

          /**
           * Get the most recently allocated, globally unique ID.
           *
           * @return  Current, globally unique ID.
           */
          long getCurrentID()
          {
             return counter;
          }

          /**
           * Prepare the next bracket of available IDs for this context.
           */
          private void prepareNextBracket()
          {
             synchronized (idContextStack)
             {
                counter = nextGlobalID;
                nextGlobalID += idBracketSize;
             }
          }
       }

       /**
        * Utility class that helps reading data from Progress exported .d files.
        * The important addition of thus class is that it attempts to read the PSC footer from
        * the file and, if found exposes the found key/values combination in form of a string map.
        */
       private static class DataFileReader
       extends FileStream
       {
          /** The PSC header magic line. This is the line that marks the start of the PSC footer. */
          private static final String PSC_HEADER_MAGIC = "PSC";

          /**
           * The standard size of the payload. If the footer is present, the last line represents the offset of
           * the PSC footer in decimal notation. It is normally 10 digits long left-padded with 0 but if the file
           * is at least 10GB large, more digits are used. 16 bytes is enough for storing file sizes up to 99TB,
           * depending on the specific EOLN terminator.
           */
          private static final int MAX_PAYLOADSIZE_SIZE = 16;

          /**
           * The maximum size of the footer. When auto detecting the PSC footer, this many bytes
           * from the end of the file will be analysed for locating the PSC footer.
           */
          private static final int MAX_PSC_FOOTER_SIZE = 1024;

          /** The map with read key/values pairs form PSC footer. */
          private Map<String, String> pscHeader = null;

          /** The code-page encoding for this file or <code>null</code> if not present. */
          private String encoding = null;

          /** The number of records that were saved to this file. */
          private long recordCount = -1;

          /** The name of the table that was exported to this file. */
          private String ldbname = null;

          /** The timestamp when this file was created.  */
          private String timestamp = null;

          /** The group separator character for numbers. */
          private char groupSeparator = 0;

          /** The decimal separator character for decimal numbers. */
          private char decimalSeparator = 0;

          /** The windowing year for date as it was set when the table was dumped. */
          private int windowingYear = 0;

          /** The date format  as it was set when the table was dumped. */
          private String dateFormat = null;

          /**
           * Constructor.
           *
           * @param   filePath
           *          File to be read.
           *
           * @throws  ErrorConditionException
           *          if there is an error finding or reading the file.
           */
          public DataFileReader(String filePath)
          throws ErrorConditionException
          {
             super(filePath, false, false, false, null);
             String fileName = new File(filePath).getName();

             try
             {
                FileStream.BOM bomId = readBOM();
                if (bomId != null)
                {
                   // overwrite the default encoding with the one detected by checking the BOM
                   encoding = bomId.name;
                   setConvertSource(encoding);
                }

                readPscFooter(fileName, (bomId != null) ? bomId.bpc : 1); // expected bytes per character
                initialize();

                // read only files shift directly into memory mapped mode at the start of the file
                // remember to skip to first character after the BOM
                map((bomId != null) ? bomId.size : 0);
             }
             catch (IOException ioe)
             {
                // This should not happen anyway.
                // If any IO issue would occur, it should have happened in the super c'tor.
                ErrorManager.recordOrThrowError(98, fileName + ": Unable to open file.");
             }
          }

          /**
           * Return the metadata value associated with this key. Each stream can have a set of
           * metadata information about how the records were exported. This may include the
           * character code-page, date-format etc.
           *
           * @param   key
           *          The key for the entry to be queried.
           *
           * @return  The derived classes will return the value for specified key, or null if this
           *          stream does not have metadata or metadata does not contain the specified key.
           *          This method always returns {@code null}.
           */
          public String getMetadata(String key)
          {
             if (key == null || pscHeader == null || pscHeader.isEmpty())
             {
                return null;
             }

             return pscHeader.get(key);
          }

          /**
           * Obtain the expected record count in this file.
           *
           * @return  The number of the record saved into this file or -1 if PSC header was not
           *          found or it does not contain the 'records' entry.
           */
          public long getRecordCount()
          {
             return recordCount;
          }

          /**
           * Don't skip a lone hyphen read from an export file.
           *
           * @return  Always {@code false}.
           */
          protected boolean skipLoneHyphenInput()
          {
             return false;
          }

          /**
           * Reads the PSC footer, storing the key/values pairs in private map pscHeader.
           *
           * @param   filename
           *          File to be read. Used only to report errors.
           * @param   bytesPerChar
           *          Standard bytes per character in the current text encoding.
           *
           * @return  The number of PSC records actually read. If negative, the footer could not
           *          be detected at the end of this file stream.
           */
          private int readPscFooter(String filename, int bytesPerChar)
          {
             if (file == null)
             {
                return -1;
             }

             String err = locateFooterStrict(bytesPerChar);
             if (err != null)
             {
                boolean err2 = !locateFooterHeuristic(bytesPerChar);
                if (err2)
                {
                   System.out.println(filename + ": " + err + " Heuristic search failed. Using defaults.");
                   return -1;
                }
                else
                {
                   // log the event but continue reading the footer
                   System.out.println(
                         filename + ": " + err + " Heuristic search found a PSC footer candidate. " +
                         "The file was probably altered after being exported from database.");
                }
             }

             // if pscHeader was already populated drop old content
             pscHeader = new HashMap<>();

             int metaRecordsCounter = 0;

             try
             {
                String keyVal = readLn();
                while (keyVal != null)
                {
                   int k = keyVal.indexOf('=');
                   if (k > 0)
                   {
                      String key = keyVal.substring(0,  k);
                      String val = keyVal.substring(k + 1);
                      pscHeader.put(key,  val);
                      ++ metaRecordsCounter;
                   }

                   // read next PSC attribute
                   keyVal = readLn();
                }
             }
             catch (EOFException eof)
             {
                // ignore, we reached the stream's end
             }
             catch (IOException | InterruptedException exc)
             {
                LOG.log(Level.FINE, "", exc);
             }

             return metaRecordsCounter;
          }

          /**
           * Attempt to locate the footer using the strict file format.
           * <p>
           * If the PSC footer is located the read file pointer is positioned right before the first PSC key/value
           * pair. Otherwise its location is undefined.
           *
           * @param   bpc
           *          The number of bytes per character in the current character encoding.
           *
           * @return  If the PSC footer is found using navigation in the strict file format the method returns
           *          {@code null}. Otherwise a non-null {@code String} is returned describing the encountered
           *          error.
           */
          private String locateFooterStrict(int bpc)
          {
             try
             {
                int pstOffsetSize = MAX_PAYLOADSIZE_SIZE * bpc;

                long wholeFileSize = file.size();
                if (wholeFileSize < pstOffsetSize)
                {
                   // file is too small to contain the offset of PSC footer
                   return "Unable to find PSC footer. File too small.";
                }

                // We don't know where the file was exported from (EOLN may be CR / CR+LF or LF).
                // So the PAYLOADSIZE_SIZE will represent:
                //    on linux:  CR + 10 x (0-9) + CR <EOF>
                //    on windows: 10 x (0-9) + CR + LF <EOF>
                // with additional digits if the file size is greater than 10GB.
                // As consequence, when going back 12 bytes back from <EOF>, on linux the first CR must be skipped
                // (bytesCountStr is empty) and we read the next line:

                // prepare to read the start of header
                map(wholeFileSize - pstOffsetSize);

                // read the bytes from the presumed start of header and decode them
                long currentPos = getPos();
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                while (available() > 0)
                {
                   baos.write(readByte());
                }
                // decode the characters
                CharBuffer buffer = baos.size() == 0 ? null : decode(baos.toByteArray());
                char[] chars = buffer == null ? new char[0] : buffer.toString().toCharArray();
                setPos(currentPos); // restore position

                int endOffset = chars.length - 1;

                // process [lastBytes]: identify the digits between last sets of EOLNs
                while (endOffset >= 0 &&
                       (chars[endOffset] == 0x0D /*CR*/ || chars[endOffset] == 0x0A /*LF*/))
                {
                   --endOffset; // drop EOLN
                }
                int startOffset = endOffset;
                while (startOffset >= 0 && chars[startOffset] >= '0' && chars[startOffset] <='9')
                {
                   --startOffset;
                }
                String bytesCountStr = new String(chars, startOffset + 1, endOffset - startOffset);

                long headerOffset = -1;
                try
                {
                   headerOffset = Long.parseLong(bytesCountStr);
                }
                catch (NumberFormatException nfe)
                {
                   // ignore the error, headerOffset remains -1 and will be handled just below
                }

                if (headerOffset == 0L)
                {
                   // auto mode: need to backtrace and look for the PSC header
                   map(Math.max(0, wholeFileSize - MAX_PSC_FOOTER_SIZE));
                   String psc = "";
                   while (!PSC_HEADER_MAGIC.equals(psc))
                   {
                      try
                      {
                         psc = readLn();
                      }
                      catch (IOException | InterruptedException e)
                      {
                         headerOffset = -1;
                         break;
                      }
                   }
                   if (headerOffset == 0)
                   {
                      headerOffset = getPos() - PSC_HEADER_MAGIC.length() - 1;
                   }
                }

                // check if offset looks correct
                if (headerOffset < 0)
                {
                   // invalid PST offset
                   return "Unable to read PSC footer: invalid PSC offset, or PSC footer not present in file.";
                }
                else if (headerOffset > wholeFileSize - MAX_PAYLOADSIZE_SIZE)
                {
                   // invalid PST offset
                   return String.format("Unable to read PSC footer at %#010x. Invalid offset.", headerOffset);
                }

                // prepare to read header now
                map(headerOffset);

                String keyVal = readLn();
                if (!PSC_HEADER_MAGIC.equals(keyVal))
                {
                   // expected PSC header of the footer,  if not found abort processing the footer
                   return String.format("Unable to read PSC footer at %#010x. Failed magic check.", headerOffset);
                }
             }
             catch (IOException | InterruptedException exc)
             {
                return exc.getMessage();
             }

             // all OK, no error
             return null;
          }

          /**
           * Attempts to locate the PSC header using heuristic methods (incrementally scan for PSC magic bytes).
           * <p>
           * If the PSC footer is located the read file pointer is positioned right before the first PSC key/value
           * pair. Otherwise its location is undefined.
           *
           * @param   bpc
           *          The number of bytes per character in the current character encoding.
           *
           * @return  {@code true} if the PSC footer was successfully found and {@code false} otherwise.
           */
          private boolean locateFooterHeuristic(int bpc)
          {
             try
             {
                long MAX_PSC_FOOTER_SIZE = 512;
                long wholeFileSize = file.size();
                long pscBufferLocation = wholeFileSize - MAX_PSC_FOOTER_SIZE * bpc;
                if (pscBufferLocation < 0)
                {
                   // adjust for really small files
                   pscBufferLocation = 0;
                }

                // prepare to read header now
                map(pscBufferLocation);

                String keyVal;
                do
                {
                   keyVal = readLn();
                   // test for the PSC header of the footer: returning success if found.
                   if (PSC_HEADER_MAGIC.equals(keyVal))
                   {
                      return true;
                   }
                   // if not found, skip to next line, until the end of stream is reached
                } while (keyVal != null);
             }
             catch (IOException | InterruptedException exc)
             {
                // ignore it, returning false anyway
             }

             return false;
          }

          /**
           * Initializes the import for this table based on the values read from the PSC footer values of this
           * stream previously read in metadata map, defaulting to values read from the configuration file.
           */
          private void initialize()
          {
             // read the timestamp. Use it for some kind of validation?
             timestamp = getMetadata("timestamp");   // eg. 2013/06/07-09:57:02

             // read the database name. Check if we are importing into the right database
             ldbname = getMetadata("ldbname");     // eg. p2j_test

             try
             {
                recordCount = Long.parseLong(getMetadata("records")); // eg. 0000000000312
             }
             catch (NumberFormatException e1)
             {
                // keep recordCount = -1
             }

             String numformat = getMetadata("numformat");    // eg. 44,46
             if (numformat == null)
             {
                // not found in PSC footer? use the default from [p2j.cfg.xml]
                numformat = defaultNumFormat;
             }
             if (numformat != null && numformat.length() >= 3 && numformat.contains(","))
             {
                StringTokenizer st = new StringTokenizer(numformat, ",", false);
                if (st.countTokens() == 2)
                {
                   try
                   {
                      groupSeparator = (char) Integer.parseInt(st.nextToken());
                      decimalSeparator = (char) Integer.parseInt(st.nextToken());

                      NumberType.setNumericFormat(String.valueOf(groupSeparator),
                                                  String.valueOf(decimalSeparator));
                   }
                   catch (NumberFormatException e)
                   {
                      // cancel set group and decimal separators
                   }
                }
             }

             String dateformat = getMetadata("dateformat");  // eg. dmy-1950
             if (dateformat == null)
             {
                // not found in PSC footer? use the default from [p2j.cfg.xml]
                dateformat = defaultDateFormat;
             }
             if (dateformat != null && dateformat.length() >= 8)
             {
                dateFormat = dateformat.substring(0, 3);
                SessionUtils.setDateFormat(dateFormat);
                try
                {
                   windowingYear = Integer.parseInt(dateformat.substring(4));
                   date.setWindowingYear(windowingYear);
                }
                catch (NumberFormatException e)
                {
                   // eat the exception, let the windowingYear field with old value
                }
             }

             encoding = getMetadata("cpstream");    // eg. ISO8859-15
             if (encoding == null)
             {
                // not found in PSC footer? use the default from [p2j.cfg.xml]
                encoding = defaultCpStream;
             }

             if (encoding != null && !encoding.equals(sourceCp) && !encoding.isEmpty())
             {
                // overwrite the current encoding with the one declared in footer
                setConvertSource(encoding);

                System.out.println(fileName + ": Set up " + encoding + " character encoding.");
             }
          }
       }

       /**
        * Auxiliary data for the word table
        */
       private static class WordTableInfo
       {
          /** Word table name */
          public final String wordTableName;

          /** field name */
          public final String fieldName;

          /** original name for denormalized fields */
          public final String originalName;

          /** flag indication that the field is case-sensitive */
          public final boolean caseSensitive;

          /** PropertyMapper for the DMO record property */
          public final PropertyMapper mapper;

          /** Field extent size */
          public final int extent;

          /** Field offset */
          public final int offset;

          /** Flag indicating that data is valid */
          public final boolean isValid;

          /**
           * Constructor.
           *
           * @param   loader
           *          The record loader.
           * @param   index
           *          The word index descriptor.
           */
          public WordTableInfo(RecordLoader loader, P2JIndex index)
          {
             boolean ok = true;
             P2JIndexComponent indexComponent = index.components().get(0);
             this.fieldName = indexComponent.getColumnName();
             this.wordTableName = index.getWordTableName();
             File dataFile = loader.getDataFile();
             String fileName = (dataFile != null) ? dataFile.getName() : "";

             if (StringUtils.isEmpty(wordTableName))
             {
                System.out.println(fileName + ": wordTableName is empty for index: [" +
                                   index.getTable() + "." + index.getName() + "]");
                ok = false;
             }

             this.caseSensitive = !indexComponent.isIgnoreCase();
             String dmoName = loader.getFullDMOName();
             DmoMeta meta = DmoMetadataManager.getDmoInfo(dmoName);
             Iterator<P2JIndex> iter = meta.getDatabaseIndexes();
             String indexName = index.getName();

             if (indexName.startsWith("idx__"))
             {
                indexName = indexName.substring("idx__".length());
             }

             while(iter.hasNext())
             {
                P2JIndex idx = iter.next();
                if (idx.getName().equals(indexName))
                {
                   indexComponent = idx.components().get(0);
                   break;
                }
             }

             this.mapper = loader.getPropertyMapperByColumn(fieldName);
             if (mapper == null)
             {
                System.out.println(fileName + ": No PropertyMapper for [" + index.getTable() + "." +
                                   fieldName + "]");
                extent = -1;
                offset  = -1;
                originalName = "";
                ok = false;
             }
             else
             {
                int ext = indexComponent.getExtent();
                extent = ext > 0 ? ext : mapper.getPropertyMeta().getOriginalExtent();
                offset = mapper.getIndex();
                originalName = extent == 0 ? "" :
                      NC.convert(mapper.getPropertyMeta().getLegacyName(), MatchPhraseConstants.TYPE_COLUMN);
             }
             isValid = ok;
          }
       }
    }