FWD » Core Development » Database

ImportWorker is responsible for creating the schema from database, creating the indexes, copy the data from the progress dump into the database, etc (ant import.db task).
The implementation for copy of data from the dump into the database uses separate INSERT statements with executeBatch API from JDBC.

When working on #11399, I have found that there is a specific PostgresSQL statement, called COPY, which is faster than SQL INSERT, optimized for bulk insert. There are also equivalents for this on MariaDB (LOAD DATA INFILE) and SQLServer (BULK INSERT).

Initial experiments for PSQL showed a throughput which has almost doubled (old: 65502 records/sec new: 114942 records/sec).

Created task branch 11467a and added an initial implementation as revision 16562.
Please note that it is unfinished (code-style and javadoc are not right), but the import has succeeded for the projects that I have managed to test.

• use bulk-copy if there is no word-index for the current table, but the dialect supports it (please explain this limitation for word indexes)
• the 'recovery' we do for batch inserts: we 'rollback' only the offending record, not the full record (like we encountered with SQLServer and mariadb I think, where a field's string value does not fit in the fixed-width char columns).
• blob/clob types which I'm not sure how they are supposed to work in 11467a
• there is no 'batching' of the records in the .d file - a .d file can be 10s of GB or more. We can not keep this in memory entirely

Constantin Asofiei wrote:

The main issues we need to solve are:

• use bulk-copy if there is no word-index for the current table, but the dialect supports it (please explain this limitation for word indexes)

Ok, I need to check this, but for normalized extends schema and for any other non-flat schema, there is a check to fallback to the old approach.

boolean doBulkCopy = dialect.supportsBulkCopy() && (!dialect.useWordTables() || wordTables.isEmpty());

Constantin Asofiei wrote:

• the 'recovery' we do for batch inserts: we 'rollback' only the offending record, not the full record (like we encountered with SQLServer and mariadb I think, where a field's string value does not fit in the fixed-width char columns).

Currently, when a record insertion fails (due to a UNIQUE constraint violation for example), it rolls back all the inserted records from that COPY statement.
In PSQL 17, there is ON-ERROR "ignore", but PSQL 14-16 does not support it.

Constantin Asofiei wrote:

• blob/clob types which I'm not sure how they are supposed to work in 11467a

There is a check for this, to fall back to the old approach. I have encountered this issue for one project.

// ImportWorker line 1676
catch (UnsupportedOperationException uoe)
{
   LOG.log(Level.INFO, String.format("%s: Bulk copy not supported for table %s: %s. Falling back to batch insert.", fileName, table, uoe.getMessage()));
   ...
}

Constantin Asofiei wrote:

• there is no 'batching' of the records in the .d file - a .d file can be 10s of GB or more. We can not keep this in memory entirely

I don't know how COPY handles transactions internally, but they are supported.

• creates an iterator which iterates through the loader instance (which loads the next record from .d) (hasNext call).
• with that iterator, we use the next record (DMO), serialize into CSV and write it to stdin stream of COPY statement.
• this stream is through that JDBC connection channel, being implemented by the PSQL JDBC driver.

Of course, these changes are not 100% right (are being generated as for POC), which will be revisited once we fix the tasks which are higher as priority.

Some measurements done on two projects, one with a smaller dump and another one, with way larger dump.

• Baseline: 5 minutes 9 seconds
• 11467/r16565: 1 minute 14 seconds

• Baseline: 41 minutes 3 seconds
• 11467/r16565: 30 minutes 15 seconds

Teodor Gorghe wrote:

I have done a couple of fixes, most of them were style fixes, support of COPY for word index tables, tuned ImportWorker to use asynchronous_commit and maintenance_work_mem.

So, in COPY mode, the triggers are not being executed, to populate the word-tables (these are only for INSERT); so you rebuild the word-tables explicitly, right?

I would have expected a greater improvement than just ~25%.

• Numeric values (e.g. BigDecimal) never contain a CSV metacharacter - this is valid only if the import worker can't set custom separators for decimal or grouping.
• how are LOBs managed?

When running ant import.db, clean.db is being executed, which drops the database.
Word-index triggers are being created in a later stage, after the data being imported to database.

CLOB data type is using COPY statement, there is no issue regarding that. I have gated BLOB data type to fallback to old batched INSERT. See the BulkCopyCsvWriter which gates the expanded extents case and the tables which has columns with BLOB data type.

The BigNumber or any Number toString implementation does not use comma as string separator (for decimal separator uses . and there is no comma grouping of three digits).

Teodor Gorghe wrote:

CLOB data type is using COPY statement, there is no issue regarding that. I have gated BLOB data type to fallback to old batched INSERT. See the BulkCopyCsvWriter which gates the expanded extents case and the tables which has columns with BLOB data type.

OK, I understand now, ImportWorker falls back to normal import if bulk fails for any reason.

The BigNumber or any Number toString implementation does not use comma as string separator (for decimal separator uses . and there is no comma grouping of three digits.

Thanks.

Constantin Asofiei wrote:

I would have expected a greater improvement than just ~25%.

I think I have a high speed SSD on my system. I see that the system runs smoother when the import process is going on.
I have a system on the office which has an HDD, it may be worth doing some comparison.

• 8755d import

100.00000% complete: 95208995 recs in 0:33:56.810 (46744/sec, 2804650/min, 168279015/hr)

• 8755d + 11467a import

100.00000% complete: 95208995 recs in 0:12:54.718 (122895/sec, 7373702/min, 442422122/hr)

So it looks really promising, on 'good data' or with just a few blob tables.

Now, do we need to validate the import (i.e. all records are imported with their correct values)?

I am finding a way to validate the import on a 41GB dump.

Teodor Gorghe wrote:

I am finding a way to validate the import on a 41GB dump.

Use our export tool to dump the content to .d and then compare the .d files to the originals.

Teodor Gorghe wrote:

Ovidiu, can you review 11467a/r16565. Thanks!

Review of 11467a / r16565

I think the update is safe. I did not spot any severe issue.

• BulkCopyCsvWriter.java:
  • lines 66-70: the full package import (.* syntax) should be used instead of individual classes;
  • lines 119/126/128: why passing the firstRecord instead of only its recordMeta directly?
• CsvRecordReader.java:
  • line 75: extends declaration should be on next line;
  • line 102: a bit strange to pass the 1st element from the 'stream' and the Iterator. The 'head' can be extracted in constructor using next(). Just noticing. No need to adjust;
  • line 129: throws declaration should be on next line;
• P2JPostgreSQLDialect.java:
  • line 2884-2996: a more intuitive way to unwrap the connection would be:

       if (conn instanceof PGConnection) return (PGConnection) conn;
       else if (conn.isWrapperFor(PGConnection.class)) return conn.unwrap(PGConnection.class);
       else throw new SQLException("failed");

• BaseRecord.java:
  • lines 172-181: the original convention Eric and I agreed upon when the BaseRecord was first created is to keep the data as encapsulated as possible and never allow direct access to it. Meanwhile, there are other 2 method that break the same rule but add some constraints. Can we use getData(PropertyMapper) instead since it is also used in import process? I really miss C++ friend classes.
  • lines 1099-1108: a similar method already exists in Record class. To use that, the ImportWorker$ImportRecordIterator should iterate on Record instead of BaseRecord;
• Persister.java:
  • line 488: I think passing the Dialect as parameter is a better solution than lookil it up, locally. (Requires adding it as parameter to Session.bulkCopy() method, as well);
• ImportWorker.java
  • line 1715: the startPos should be always 0, but having more flexibility will not hurt;
  • line 2159-2160: The parameter list should be chopped with one parameter on each line if the method declaration line is longer than 110 chars;
  • line 4303 &4363: missing method javadoc.

Greg Shah wrote:

Use our export tool to dump the content to .d and then compare the .d files to the originals.

I have imported and then exported the dump.
I see just one difference, the exported dump is encoded in a different charset.

That is expected since the dump as imported would be in 8859-1 and the database as imported is UTF-8. The import transforms the data between charsets.

I think that is a pass.
However, if you want to do a cross-check you can do a SQL database dump (backup) in text (SQL) format and compare the result for a database imported with the trunk and same database imported with 11467a. These dumps show the full SQL database content, including schema and the UDFs.

SQL dump gives output difference every time when you import from database, due to the fact how data is being stored in the PG heap. Recid is also different.

I have done a comparison by stripping recids and sorting the records inside COPY blocks, and I don't see any difference (word tables are also populated right).

Addressed code review #11467-17 in 11467a/r16566.

Teodor, please also test a (small) import with 10614_main.

Constantin Asofiei wrote:

Teodor, please also test a (small) import with 10614_main.

On that app, we don't have the .d files, do I you mean to extract .d files and then reimport?

Teodor Gorghe wrote:

Constantin Asofiei wrote:

Teodor, please also test a (small) import with 10614_main.

On that app, we don't have the .d files, do I you mean to extract .d files and then reimport?

That, or use 10614_main to convert the .df schema for another app and import with it (you don't have to fully convert or parse the code).

Let me know if and when the branch is ready to merge.

Currently working on #7143-1977, but I think it will be ready EOD.

Tested 10614_main to import the database. Import time has reduced with nearly 50%.

• Fixed the NPE when COPY fails and rollbacks to batch insert.

I see this from claude, it may be a valid point:

• [MINOR] functional ImportWorker.importTable (bulk path): Column-less ("empty") tables — a DMO with only the surrogate PK, a first-class case that RecordLoader.isRecordEmpty() explicitly supports — with a PSC-footer count > 0 crash the bulk path. iterator.hasNext() returns true (empty records are non-null), so new BulkCopyCsvWriter(loader.getMetadata(), loader.getProperties().get(0)) runs and .get(0) on the empty properties list throws IndexOutOfBoundsException. It bypasses the catch (UnsupportedOperationException) and lands in catch (Exception), logged at INFO as a misleading "Bulk copy failed", so every such table silently loses the COPY optimization. No data loss (batch fallback handles it correctly). Fix: guard with if (!loader.getProperties().isEmpty()) (equivalently !loader.isRecordEmpty()) and let the batch path handle column-less tables — note a lazy-witness fix does not work, as such a table has no PropertyMapper and appendRow calls dmo.getData(witness) which requires a non-null mapper.

Done in 11467a/r16568.

Teodor Gorghe wrote:

Done in 11467a/r16568.

Ovidiu, please review again.

Teodor Gorghe wrote:

SQL dump gives output difference every time when you import from database, due to the fact how data is being stored in the PG heap. Recid is also different.

I haven't reviewed the code, so this may not be a relevant inquiry, but please confirm:

• How are primary keys being generated/assigned now?
• Are primary keys still globally unique across all tables?
• At the end of the import, whether tables have used the COPY mode or batched INSERT mode, is the p2j_id_generator_sequence sequence still being properly updated to reflect the highest assigned primary key? We rely on this at runtime when creating new records.

Eric Faulhaber wrote:

I haven't reviewed the code, so this may not be a relevant inquiry, but please confirm:

• How are primary keys being generated/assigned now?

Using the same IDContext instance like how to batch insert path works.

• Obtaining current id and passing to the iterator: codecode

                 if (doBulkCopy)
                 {
                    long savedId = idCtx.getCurrentID();
                    long savedRest = idCtx.getRestInBucket();
                    session = openSession();
                    ImportRecordIterator iterator = new ImportRecordIterator(loader,
                                                                             stream,
                                                                             idCtx,
                                                                             notNullProperties,
                                                                             hasNotNull,
                                                                             fileName,
                                                                             expectedRecords);
  

• ImportRecordIterator hasNext(): codecode

        @Override
        public boolean hasNext()
        {
           ...
           nextRecord.primaryKey(idCtx.getNextID());
           // No overflow handling here: it applies only to SQL Server (see handleOverflowingData),
           // which does not support bulk copy, so on this PostgreSQL-only path it is always a no-op.
  
           count++;
           return true;
        }
  

• When an exception happens, calls discardBulkCopyAttempt which resets the context: section 1section 1

                    catch (UnsupportedOperationException uoe)
                    {
                       LOG.log(Level.INFO, String.format("%s: Bulk copy not supported for table %s: %s. " +
                                           "Falling back to batch insert.", fileName, table, uoe.getMessage()));
                       stream = discardBulkCopyAttempt(session, stream, dataFile, idCtx, savedId, savedRest);
                       session = null;
                       // Carry forward the footer state so the batch pass does not repeat any warning.
                       expectedRecords = iterator.getExpectedRecords();
                       doBulkCopy = false;
                    }
                    catch (Exception exc)
                    {
                       // A data error (bad datum, constraint or unique-index violation) aborts the whole
                       // atomic COPY; fall back to the batch path, which drops just the offending record(s)
                       // and reports fatal only if it, too, cannot complete.
                       LOG.log(Level.INFO, String.format("%s: Bulk copy failed for table %s (%s). " +
                                           "Falling back to batch insert.", fileName, table, exc.getMessage()));
                       LOG.log(Level.FINE, "Bulk copy failure detail", exc);
                       stream = discardBulkCopyAttempt(session, stream, dataFile, idCtx, savedId, savedRest);
                       session = null;
                       // Carry forward the footer state so the batch pass does not repeat any warning.
                       expectedRecords = iterator.getExpectedRecords();
                       doBulkCopy = false;
                    }
  

  discardBulkCopyAttemptdiscardBulkCopyAttempt

        private DataFileReader discardBulkCopyAttempt(Session session,
                                                      DataFileReader stream,
                                                      File dataFile,
                                                      IDContext idCtx,
                                                      long savedId,
                                                      long savedRest)
        throws ErrorConditionException
        {
           if (session != null && session.isOpen())
           {
              try
              {
                 session.rollback();
              }
              catch (Exception exc)
              {
                 LOG.log(Level.FINE, "Error rolling back bulk-copy session", exc);
              }
  
              try
              {
                 session.close();
              }
              catch (Exception exc)
              {
                 LOG.log(Level.FINE, "Error closing bulk-copy session", exc);
              }
           }
  
           stream.close();
           DataFileReader fresh = new DataFileReader(dataFile.getPath());
           if (mapBytes != null)
           {
              fresh.setInvalidBytesMapping(mapBytes);
           }
  
           idCtx.resetTo(savedId, savedRest);
           return fresh;
        }
  

• Are primary keys still globally unique across all tables?

• Yes, we use the same IDContext/TenantIDContext, which handles the primary key in brackets, across multiple threads.
• Please take in consideration that the database rollback in PSQL does not restore the sequence value (is non-transactional). This may be reason why we handle the primary key in memory and we create the sequence at the end.

• At the end of the import, whether tables have used the COPY mode or batched INSERT mode, is the p2j_id_generator_sequence sequence still being properly updated to reflect the highest assigned primary key? We rely on this at runtime when creating new records.

• Yes, of course. It still uses last id of IDContext to create the sequence: codecode

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

• table has records more than the number of IDContext bracket size and the import process is multi-threaded.
• when a rollback happens and there is a need to restore the IDContext, I have added a safe-guard to restore the IDContext only when they were inserted records which does not exceed the first current bracket.
• when the first current bracket is exceeded, IDContext.resetTo is essentially a NO-OP.
• why works in this way: I thought that it is better to leave a gap between recids, than the unnecessary work to restore the entire IDContext.

Constantin Asofiei wrote:

Eric, any other concerns?

Let's be sure to include multi-tenant in the testing.

Other than that, I just wanted to make sure we were good on the primary keys and it seems we are. Thanks for the answers, Teodor.

Let's try to get this into trunk ASAP after Ovidiu's (hopefully) final review and any other testing that needs to be done.