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;
}
}
}