importdelimitedworker.java

eq跨平台查询工具源码 eq跨平台查询工具源码

                                }

                            }

                        }
                        
                        if (isBatch) {
                            prepStmnt.addBatch();
                        } else {
                            int result = prepStmnt.executeUpdate();
                            tableInsertCount += result;
                            commitCount += result;
                        }

                        rollbackCount++;
                        // check the rollback segment
                        if (rollbackCount == rollbackSize) {
                            if (isBatch) {
                                int result = getBatchResult(prepStmnt.executeBatch())[0];
                                tableInsertCount += result;
                                commitCount += result;
                                prepStmnt.clearBatch();
                            }
                            conn.commit();
                            totalInsertCount += commitCount;
                            tableCommitCount = tableInsertCount;
                            rollbackCount = 0;
                            commitCount = 0;
                        }

                        // reset bound variables
                        if (hasColumnNames) {
                            for (int k = 0; k < columnCount; k++) {
                                boundVariables.put(columns.get(k), VARIABLE_NOT_BOUND);
                            }
                        }

                    }
                    catch (SQLException e) {
                        logException(e);
                        errorCount++;

                        if (!isBatch) {
                            outputBuffer.append("Error inserting data from line ");
                            outputBuffer.append(lineNumber);
                            outputExceptionError(null, e);
                        }
                        else {
                            outputBuffer.append("Error on last batch execution");
                            outputExceptionError(null, e);
                        }

                        if (haltOnError) {
                            processResult = FAILED;
                            conn.rollback();
                            getParent().cancelTransfer();
                            throw new InterruptedException();
                        }

                    }

                }

                // ----------------------------
                // file/table has ended here
                
                if (isBatch) {
                    
                    int[] batchResult = null;
                    
                    try {
                        batchResult = getBatchResult(prepStmnt.executeBatch());
                        int result = batchResult[0];
                        tableInsertCount += result;
                        commitCount += result;
                        tableCommitCount = tableInsertCount;
                    }
                    catch (BatchUpdateException e) {
                        logException(e);
                        int[] updateCounts = e.getUpdateCounts();
                        batchResult = getBatchResult(updateCounts);
                        errorCount += batchResult[1];
                        if (errorCount == 0) {
                            errorCount = 1;
                        }

                        outputBuffer.append("An error occured during the batch process: ");
                        outputBuffer.append(e.getMessage());
                        
                        SQLException _e = e.getNextException();
                        while (_e != null) {
                            outputBuffer.append("\nNext Exception: ");
                            outputBuffer.append(_e.getMessage());
                            _e = _e.getNextException();
                        }
                        
                        outputBuffer.append("\n\nRecords processed to the point ");
                        outputBuffer.append("where this error occurred: ");
                        outputBuffer.append(updateCounts.length);
                        appendProgressErrorText(outputBuffer);
                        processResult = FAILED;
                    }
                    
                    //  Log.debug("commitCount: " + commitCount + 
                    //                      " batch: " + batchResult[0]);
                    
                    if (tableRowCount != tableInsertCount) {
                        conn.rollback();
                        
                        if (onError == ImportExportProcess.STOP_TRANSFER) {
                            getParent().cancelTransfer();
                            processResult = FAILED;
                            throw new InterruptedException();
                        }
                        
                    }
                    
                }

                boolean doCommit = true;
                if (failed && !isBatch && 
                        rollbackSize != ImportExportProcess.COMMIT_END_OF_ALL_FILES) {
                    
                    int yesNo = GUIUtilities.displayYesNoDialog(
                                    "The process completed with errors.\n" +
                                    "Do you wish to commit the last block?",
                                    "Confirm commit");
                    
                    doCommit = (yesNo == JOptionPane.YES_OPTION);
                }

                // do the commit if ok from above
                // and if rollback size selected is end of file
                if (rollbackSize == ImportExportProcess.COMMIT_END_OF_FILE) {
                    if (doCommit) {
                        conn.commit();
                        totalInsertCount += commitCount;
                        tableCommitCount = tableInsertCount;
                        commitCount = 0;
                    } else {
                        conn.rollback();
                    }
                }
                
                reader.close();
                fileReader.close();
                prepStmnt.close();
                
                // update the progress display
                printTableResult(tableRowCount, 
                        tableInsertCount, dto.getTableName());
                setProgressStatus(100);
                
                // reset the checks
                hasDate = false;
                failed = false;
                
            }
            
            // commit the last remaining block or where 
            // set to commit at the end of all files
            if (rollbackSize != ImportExportProcess.COMMIT_END_OF_FILE) {
                setProgressStatus(100);
                boolean doCommit = true;
                if (errorCount > 0 && errorCount != totalRecordCount) {
                    int yesNo = GUIUtilities.displayYesNoDialog(
                                    "The process completed with errors.\n" +
                                    "Do you wish to commit the changes?",
                                    "Confirm commit");
                    doCommit = (yesNo == JOptionPane.YES_OPTION);
                }
                
                if (doCommit) {
                    conn.commit();
                    totalInsertCount += commitCount;
                } else {
                    conn.rollback();
                }
                
            }

            processResult = SUCCESS;
        }
        catch (InterruptedException e) {
            
            if (processResult != FAILED) {
                processResult = CANCELLED;
            }

            try {
                if (prepStmnt != null) {
                    prepStmnt.cancel();
                }
                if (conn != null) {
                    conn.rollback();
                }
            } catch (SQLException e2) {
                outputExceptionError("Error rolling back transaction", e);
            }

        }
        catch (Exception e) {
            logException(e);
            outputBuffer.append("Error processing data from line ");
            outputBuffer.append(lineNumber);
            outputExceptionError("\nUnrecoverable error importing table data from file", e);

            int yesNo = GUIUtilities.displayYesNoDialog(
                            "The process encountered errors.\n" +
                            "Do you wish to commit the last transaction block?",
                            "Confirm commit");
            boolean doCommit = (yesNo == JOptionPane.YES_OPTION);
            
            try {
                if (doCommit) {
                    conn.commit();
                    totalInsertCount += commitCount;
                } else {
                    conn.rollback();
                }
            }
            catch (SQLException e2) {
                logException(e2);
                outputExceptionError("Error processing last transaction block", e2);
            }
            processResult = FAILED;
        }
        finally {
            finish();
            releaseResources(getParent().getDatabaseConnection());
            
            if (totalRecordCount == 0 || errorCount > 0) {
                processResult = FAILED;
            }
            
            setTableCount(tableCount);
            setRecordCount(totalRecordCount);
            setRecordCountProcessed(totalInsertCount);
            setErrorCount(errorCount);
            
            setProgressStatus(100);
            GUIUtilities.scheduleGC();
        }

        return processResult;        
    }
    
    private void logException(Throwable e) {
        if (Log.isDebugEnabled()) {
            Log.debug("Error on delimited import.", e);
        }
    }

    private boolean isIndexIgnored(List<Integer> ignoredIndexes, int index) {
        for (int i = 0, n = ignoredIndexes.size(); i < n; i++) {
            if (index == ignoredIndexes.get(i).intValue()) {
                return true;
            }
        }
        return false;
    }
    
    private int getIndexOfColumn(Vector<ColumnData> columns, ColumnData cd) {
        for (int i = 0, n = columns.size(); i < n; i++) {
            if (columns.get(i) == cd) {
                return i;
            }
        }
        return -1;
    }
    
    /**
     * Prints the table specific execution results to the output buffer.
     */
    private void printTableResult(int tableRowCount, 
                                  int tableInsertCount, 
                                  String tableName) {
        // update the progress display
        outputBuffer.append("Records processed: ");
        outputBuffer.append(tableRowCount);
        outputBuffer.append("\nRecords inserted: ");
        outputBuffer.append(tableInsertCount);
        outputBuffer.append("\nImport complete for table: ");
        outputBuffer.append(tableName);
        appendProgressText(outputBuffer);
    }

    
    private int[] getBatchResult(int[] updateCount) {
        int insert = 0;
        int success = 0;
        int errors = 0;
        
        // annoying as becoming a little db specific,
        // but Oracle returns -2 on a successful batch
        // execution using prepared statement (-3 on error)
        ConnectionDataSource cds = (ConnectionDataSource)
                ConnectionManager.getDataSource(getParent().getDatabaseConnection());
        if (cds.isUsingOracleThinDriver()) {
            success = -2;
        } else {
            success = 1;
        }
        
        for (int i = 0; i < updateCount.length; i++) {
            if (updateCount[i] == success) {
                insert++;
            } else {
                errors++;
            }
        }
        
        int[] result = {insert, errors};
        return result;
    }
    
    /**
     * Cancels the current in-process transfer. 
     */
    public void cancelTransfer() {
        worker.interrupt();
        getParent().cancelTransfer();
    }
    
    /** 
     * Indicates that the process has completed. 
     */
    public void finished() {}
    
}