private void createTableFor()

in oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java [1284:1416]
106 lines of code
23 McCabe index (conditional complexity)

    private void createTableFor(Connection con, Collection<? extends Document> col, RDBTableMetaData tmd, List<String> tablesCreated,
            List<String> tablesPresent, int initialSchema, int upgradeToSchema) throws SQLException {

        String dbname = this.dbInfo.toString();
        if (con.getMetaData().getURL() != null) {
            dbname += " (" + con.getMetaData().getURL() + ")";
        }
        String tableName = tmd.getName();

        Statement checkStatement = null;

        ResultSet checkResultSet = null;
        Statement creatStatement = null;

        try {
            // avoid PreparedStatement due to weird DB2 behavior (OAK-6237)
            checkStatement = con.createStatement();
            checkResultSet = checkStatement.executeQuery("select * from " + tableName + " where ID = '0'");

            // try to discover size of DATA column and binary-ness of ID
            ResultSetMetaData met = checkResultSet.getMetaData();
            obtainFlagsFromResultSetMeta(met, tmd);

            // check that all required columns are present
            Set<String> requiredColumns = new HashSet<String>(REQUIREDCOLUMNS);
            Set<String> unknownColumns = new HashSet<String>();
            boolean hasVersionColumn = false;
            boolean hasSDTypeColumn = false;
            for (int i = 1; i <= met.getColumnCount(); i++) {
                String cname = met.getColumnName(i).toLowerCase(Locale.ENGLISH);
                if (!requiredColumns.remove(cname)) {
                    if (!OPTIONALCOLUMNS.contains(cname)) {
                        unknownColumns.add(cname);
                    }
                }
                if (cname.equals("version")) {
                    hasVersionColumn = true;
                }
                if (cname.equals("sdtype")) {
                    hasSDTypeColumn = true;
                }
            }

            if (!requiredColumns.isEmpty()) {
                String message = String.format("Table %s: the following required columns are missing: %s", tableName,
                        requiredColumns.toString());
                LOG.error(message);
                throw new DocumentStoreException(message);
            }

            if (!unknownColumns.isEmpty()) {
                String message = String.format("Table %s: the following columns are unknown and will not be maintained: %s",
                        tableName, unknownColumns.toString());
                LOG.info(message);
            }

            String tableInfo = RDBJDBCTools.dumpResultSetMeta(met);
            tmd.setSchemaInfo(tableInfo);
            Set<String> indexOn = new HashSet<String>();
            String indexInfo = dumpIndexData(con.getMetaData(), met, tableName, indexOn);
            tmd.setIndexInfo(indexInfo);

            closeResultSet(checkResultSet);
            boolean dbWasChanged = false;

            if (this.readOnly) {
                LOG.debug("Skipping table update code because store is initialized in readOnly mode");
            }
            else {
                if (!hasVersionColumn && upgradeToSchema >= 1) {
                    dbWasChanged |= upgradeTable(con, tableName, 1);
                }

                if (!hasSDTypeColumn && upgradeToSchema >= 2) {
                    dbWasChanged |= upgradeTable(con, tableName, 2);
                }

                if (!indexOn.contains("MODIFIED") && col == Collection.NODES) {
                    dbWasChanged |= addModifiedIndex(con, tableName);
                }
            }

            tablesPresent.add(tableName);

            if (dbWasChanged) {
                getTableMetaData(con, col, tmd);
            }
        } catch (SQLException ex) {
            // table does not appear to exist
            con.rollback();

            LOG.debug("trying to read from '" + tableName + "'", ex);
            if (this.readOnly) {
                throw new SQLException("Would like to create table '" + tableName
                        + "', but RDBDocumentStore has been initialized in 'readonly' mode");
            }

            try {
                creatStatement = con.createStatement();
                creatStatement.execute(this.dbInfo.getTableCreationStatement(tableName, initialSchema));
                creatStatement.close();

                for (String ic : this.dbInfo.getIndexCreationStatements(tableName, initialSchema)) {
                    creatStatement = con.createStatement();
                    creatStatement.execute(ic);
                    creatStatement.close();
                }

                con.commit();

                if (initialSchema < 1 && upgradeToSchema >= 1) {
                    upgradeTable(con, tableName, 1);
                }

                if (initialSchema < 2 && upgradeToSchema >= 2) {
                    upgradeTable(con, tableName, 2);
                }

                tablesCreated.add(tableName);

                getTableMetaData(con, col, tmd);
            }
            catch (SQLException ex2) {
                LOG.error("Failed to create table '" + tableName + "' in '" + dbname + "'", ex2);
                throw ex2;
            }
        }
        finally {
            closeResultSet(checkResultSet);
            closeStatement(checkStatement);
            closeStatement(creatStatement);
        }
    }