001/*
002 * (C) Copyright 2006-2011 Nuxeo SA (http://nuxeo.com/) and others.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 *
016 * Contributors:
017 *     Olivier Grisel
018 *     Florent Guillaume
019 */
020package org.nuxeo.ecm.directory.sql;
021
022import java.io.Serializable;
023import java.sql.Connection;
024import java.sql.PreparedStatement;
025import java.sql.ResultSet;
026import java.sql.SQLException;
027import java.util.Arrays;
028import java.util.Collections;
029import java.util.HashSet;
030import java.util.LinkedList;
031import java.util.List;
032import java.util.Map;
033import java.util.Set;
034import java.util.function.Consumer;
035
036import org.nuxeo.ecm.core.schema.types.SchemaImpl;
037import org.nuxeo.ecm.core.schema.types.primitives.StringType;
038import org.nuxeo.ecm.core.storage.sql.ColumnType;
039import org.nuxeo.ecm.core.storage.sql.jdbc.db.Column;
040import org.nuxeo.ecm.core.storage.sql.jdbc.db.Delete;
041import org.nuxeo.ecm.core.storage.sql.jdbc.db.Insert;
042import org.nuxeo.ecm.core.storage.sql.jdbc.db.Select;
043import org.nuxeo.ecm.core.storage.sql.jdbc.db.Table;
044import org.nuxeo.ecm.core.storage.sql.jdbc.db.Table.IndexType;
045import org.nuxeo.ecm.core.storage.sql.jdbc.dialect.Dialect;
046import org.nuxeo.ecm.directory.AbstractReference;
047import org.nuxeo.ecm.directory.BaseDirectoryDescriptor;
048import org.nuxeo.ecm.directory.Directory;
049import org.nuxeo.ecm.directory.DirectoryCSVLoader;
050import org.nuxeo.ecm.directory.DirectoryException;
051import org.nuxeo.ecm.directory.ReferenceDescriptor;
052import org.nuxeo.ecm.directory.Session;
053
054public class TableReference extends AbstractReference {
055
056    protected String tableName;
057
058    protected String sourceColumn;
059
060    protected String targetColumn;
061
062    protected String dataFileName;
063
064    private Table table;
065
066    private Dialect dialect;
067
068    private boolean initialized = false;
069
070    /**
071     * @since 9.2
072     */
073    public TableReference(String fieldName, String directory, String tableName, String sourceColumn,
074            String targetColumn, String dataFileName) {
075        super(fieldName, directory);
076        this.tableName = tableName;
077        this.sourceColumn = sourceColumn;
078        this.targetColumn = targetColumn;
079        this.dataFileName = dataFileName;
080    }
081
082    /**
083     * @since 9.2
084     */
085    public TableReference(TableReferenceDescriptor descriptor) {
086        this(descriptor.getFieldName(), descriptor.getTargetDirectoryName(), descriptor.getTableName(),
087                descriptor.getSourceColumn(), descriptor.getTargetColumn(), descriptor.getDataFileName());
088    }
089
090    /**
091     * @since 9.2
092     */
093    public TableReference(ReferenceDescriptor descriptor) {
094        this(descriptor.getFieldName(), descriptor.getDirectory(), descriptor.getReferenceName(),
095                descriptor.getSource(), descriptor.getTarget(), descriptor.getDataFileName());
096    }
097
098    private SQLDirectory getSQLSourceDirectory() throws DirectoryException {
099        Directory dir = getSourceDirectory();
100        return (SQLDirectory) dir;
101    }
102
103    private void initialize(SQLSession sqlSession) throws DirectoryException {
104        Connection connection = sqlSession.sqlConnection;
105        SQLDirectory directory = getSQLSourceDirectory();
106        Table table = getTable();
107        SQLHelper helper = new SQLHelper(connection, table, directory.getDescriptor().getCreateTablePolicy());
108        boolean loadData = helper.setupTable();
109        if (loadData && dataFileName != null) {
110            // fake schema for DirectoryCSVLoader.loadData
111            SchemaImpl schema = new SchemaImpl(tableName, null);
112            schema.addField(sourceColumn, StringType.INSTANCE, null, 0, Collections.emptySet());
113            schema.addField(targetColumn, StringType.INSTANCE, null, 0, Collections.emptySet());
114            Insert insert = new Insert(table);
115            for (Column column : table.getColumns()) {
116                insert.addColumn(column);
117            }
118            try (PreparedStatement ps = connection.prepareStatement(insert.getStatement())) {
119                Consumer<Map<String, Object>> loader = new Consumer<Map<String, Object>>() {
120                    @Override
121                    public void accept(Map<String, Object> map) {
122                        try {
123                            ps.setString(1, (String) map.get(sourceColumn));
124                            ps.setString(2, (String) map.get(targetColumn));
125                            ps.execute();
126                        } catch (SQLException e) {
127                            throw new DirectoryException(e);
128                        }
129                    }
130                };
131                DirectoryCSVLoader.loadData(dataFileName, BaseDirectoryDescriptor.DEFAULT_DATA_FILE_CHARACTER_SEPARATOR,
132                        schema, loader);
133            } catch (SQLException e) {
134                throw new DirectoryException(String.format("Table '%s' initialization failed", tableName), e);
135            }
136        }
137    }
138
139    @Override
140    public void addLinks(String sourceId, List<String> targetIds) throws DirectoryException {
141        if (targetIds == null) {
142            return;
143        }
144        try (SQLSession session = getSQLSession()) {
145            addLinks(sourceId, targetIds, session);
146        }
147    }
148
149    @Override
150    public void addLinks(List<String> sourceIds, String targetId) throws DirectoryException {
151        if (sourceIds == null) {
152            return;
153        }
154        try (SQLSession session = getSQLSession()) {
155            addLinks(sourceIds, targetId, session);
156        }
157    }
158
159    @Override
160    public void addLinks(String sourceId, List<String> targetIds, Session session) throws DirectoryException {
161        if (targetIds == null) {
162            return;
163        }
164        SQLSession sqlSession = (SQLSession) session;
165        maybeInitialize(sqlSession);
166        for (String targetId : targetIds) {
167            addLink(sourceId, targetId, sqlSession, true);
168        }
169    }
170
171    @Override
172    public void addLinks(List<String> sourceIds, String targetId, Session session) throws DirectoryException {
173        if (sourceIds == null) {
174            return;
175        }
176        SQLSession sqlSession = (SQLSession) session;
177        maybeInitialize(sqlSession);
178        for (String sourceId : sourceIds) {
179            addLink(sourceId, targetId, sqlSession, true);
180        }
181    }
182
183    public boolean exists(String sourceId, String targetId, SQLSession session) throws DirectoryException {
184        // String selectSql = String.format(
185        // "SELECT COUNT(*) FROM %s WHERE %s = ? AND %s = ?", tableName,
186        // sourceColumn, targetColumn);
187
188        Table table = getTable();
189        Select select = new Select(table);
190        select.setFrom(table.getQuotedName());
191        select.setWhat("count(*)");
192        String whereString = String.format("%s = ? and %s = ?", table.getColumn(sourceColumn).getQuotedName(),
193                table.getColumn(targetColumn).getQuotedName());
194
195        select.setWhere(whereString);
196
197        String selectSql = select.getStatement();
198        if (session.logger.isLogEnabled()) {
199            session.logger.logSQL(selectSql, Arrays.<Serializable> asList(sourceId, targetId));
200        }
201
202        PreparedStatement ps = null;
203        ResultSet rs = null;
204        try {
205            ps = session.sqlConnection.prepareStatement(selectSql);
206            ps.setString(1, sourceId);
207            ps.setString(2, targetId);
208            rs = ps.executeQuery();
209            rs.next();
210            return rs.getInt(1) > 0;
211        } catch (SQLException e) {
212            throw new DirectoryException(String.format("error reading link from %s to %s", sourceId, targetId), e);
213        } finally {
214            try {
215                if (rs != null) {
216                    rs.close();
217                }
218                if (ps != null) {
219                    ps.close();
220                }
221            } catch (SQLException sqle) {
222                throw new DirectoryException(sqle);
223            }
224        }
225    }
226
227    public void addLink(String sourceId, String targetId, SQLSession session, boolean checkExisting)
228            throws DirectoryException {
229        // OG: the following query should have avoided the round trips but
230        // does not work for some reason that might be related to a bug in the
231        // JDBC driver:
232        //
233        // String sql = String.format(
234        // "INSERT INTO %s (%s, %s) (SELECT ?, ? FROM %s WHERE %s = ? AND %s =
235        // ? HAVING COUNT(*) = 0)", tableName, sourceColumn, targetColumn,
236        // tableName, sourceColumn, targetColumn);
237
238        // first step: check that this link does not exist yet
239        if (checkExisting && exists(sourceId, targetId, session)) {
240            return;
241        }
242
243        // second step: add the link
244
245        // String insertSql = String.format(
246        // "INSERT INTO %s (%s, %s) VALUES (?, ?)", tableName,
247        // sourceColumn, targetColumn);
248        Table table = getTable();
249        Insert insert = new Insert(table);
250        insert.addColumn(table.getColumn(sourceColumn));
251        insert.addColumn(table.getColumn(targetColumn));
252        String insertSql = insert.getStatement();
253        if (session.logger.isLogEnabled()) {
254            session.logger.logSQL(insertSql, Arrays.<Serializable> asList(sourceId, targetId));
255        }
256
257        PreparedStatement ps = null;
258        try {
259            ps = session.sqlConnection.prepareStatement(insertSql);
260            ps.setString(1, sourceId);
261            ps.setString(2, targetId);
262            ps.execute();
263        } catch (SQLException e) {
264            throw new DirectoryException(String.format("error adding link from %s to %s", sourceId, targetId), e);
265        } finally {
266            try {
267                if (ps != null) {
268                    ps.close();
269                }
270            } catch (SQLException sqle) {
271                throw new DirectoryException(sqle);
272            }
273        }
274    }
275
276    protected List<String> getIdsFor(String valueColumn, String filterColumn, String filterValue)
277            throws DirectoryException {
278        try (SQLSession session = getSQLSession()) {
279            // String sql = String.format("SELECT %s FROM %s WHERE %s = ?",
280            // table.getColumn(valueColumn), tableName, filterColumn);
281            Table table = getTable();
282            Select select = new Select(table);
283            select.setWhat(table.getColumn(valueColumn).getQuotedName());
284            select.setFrom(table.getQuotedName());
285            select.setWhere(table.getColumn(filterColumn).getQuotedName() + " = ?");
286
287            String sql = select.getStatement();
288            if (session.logger.isLogEnabled()) {
289                session.logger.logSQL(sql, Collections.<Serializable> singleton(filterValue));
290            }
291
292            List<String> ids = new LinkedList<String>();
293            try (PreparedStatement ps = session.sqlConnection.prepareStatement(sql)) {
294                ps.setString(1, filterValue);
295                try (ResultSet rs = ps.executeQuery()) {
296                    while (rs.next()) {
297                        ids.add(rs.getString(valueColumn));
298                    }
299                    return ids;
300                }
301            } catch (SQLException e) {
302                throw new DirectoryException("error fetching reference values: ", e);
303            }
304        }
305    }
306
307    @Override
308    public List<String> getSourceIdsForTarget(String targetId) throws DirectoryException {
309        return getIdsFor(sourceColumn, targetColumn, targetId);
310    }
311
312    @Override
313    public List<String> getTargetIdsForSource(String sourceId) throws DirectoryException {
314        return getIdsFor(targetColumn, sourceColumn, sourceId);
315    }
316
317    public void removeLinksFor(String column, String entryId, SQLSession session) throws DirectoryException {
318        Table table = getTable();
319        String sql = String.format("DELETE FROM %s WHERE %s = ?", table.getQuotedName(), table.getColumn(column)
320                                                                                              .getQuotedName());
321        if (session.logger.isLogEnabled()) {
322            session.logger.logSQL(sql, Collections.<Serializable> singleton(entryId));
323        }
324        PreparedStatement ps = null;
325        try {
326            ps = session.sqlConnection.prepareStatement(sql);
327            ps.setString(1, entryId);
328            ps.execute();
329        } catch (SQLException e) {
330            throw new DirectoryException("error remove links to " + entryId, e);
331        } finally {
332            try {
333                if (ps != null) {
334                    ps.close();
335                }
336            } catch (SQLException sqle) {
337                throw new DirectoryException(sqle);
338            }
339        }
340    }
341
342    @Override
343    public void removeLinksForSource(String sourceId, Session session) throws DirectoryException {
344        SQLSession sqlSession = (SQLSession) session;
345        maybeInitialize(sqlSession);
346        removeLinksFor(sourceColumn, sourceId, sqlSession);
347    }
348
349    @Override
350    public void removeLinksForTarget(String targetId, Session session) throws DirectoryException {
351        SQLSession sqlSession = (SQLSession) session;
352        maybeInitialize(sqlSession);
353        removeLinksFor(targetColumn, targetId, sqlSession);
354    }
355
356    @Override
357    public void removeLinksForSource(String sourceId) throws DirectoryException {
358        try (SQLSession session = getSQLSession()) {
359            removeLinksForSource(sourceId, session);
360        }
361    }
362
363    @Override
364    public void removeLinksForTarget(String targetId) throws DirectoryException {
365        try (SQLSession session = getSQLSession()) {
366            removeLinksForTarget(targetId, session);
367        }
368    }
369
370    public void setIdsFor(String idsColumn, List<String> ids, String filterColumn, String filterValue,
371            SQLSession session) throws DirectoryException {
372
373        List<String> idsToDelete = new LinkedList<String>();
374        Set<String> idsToAdd = new HashSet<String>();
375        if (ids != null) { // ids may be null
376            idsToAdd.addAll(ids);
377        }
378        Table table = getTable();
379
380        // iterate over existing links to find what to add and what to remove
381        String selectSql = String.format("SELECT %s FROM %s WHERE %s = ?", table.getColumn(idsColumn).getQuotedName(),
382                table.getQuotedName(), table.getColumn(filterColumn).getQuotedName());
383        PreparedStatement ps = null;
384        try {
385            ps = session.sqlConnection.prepareStatement(selectSql);
386            ps.setString(1, filterValue);
387            ResultSet rs = ps.executeQuery();
388            while (rs.next()) {
389                String existingId = rs.getString(1);
390                if (idsToAdd.contains(existingId)) {
391                    // to not add already existing ids
392                    idsToAdd.remove(existingId);
393                } else {
394                    // delete unwanted existing ids
395                    idsToDelete.add(existingId);
396                }
397            }
398            rs.close();
399        } catch (SQLException e) {
400            throw new DirectoryException("failed to fetch existing links for " + filterValue, e);
401        } finally {
402            try {
403                if (ps != null) {
404                    ps.close();
405                }
406            } catch (SQLException sqle) {
407                throw new DirectoryException(sqle);
408            }
409        }
410
411        if (!idsToDelete.isEmpty()) {
412            // remove unwanted links
413
414            // String deleteSql = String.format(
415            // "DELETE FROM %s WHERE %s = ? AND %s = ?", tableName,
416            // filterColumn, idsColumn);
417            Delete delete = new Delete(table);
418            String whereString = String.format("%s = ? AND %s = ?", table.getColumn(filterColumn).getQuotedName(),
419                    table.getColumn(idsColumn).getQuotedName());
420            delete.setWhere(whereString);
421            String deleteSql = delete.getStatement();
422
423            try {
424                ps = session.sqlConnection.prepareStatement(deleteSql);
425                for (String unwantedId : idsToDelete) {
426                    if (session.logger.isLogEnabled()) {
427                        session.logger.logSQL(deleteSql, Arrays.<Serializable> asList(filterValue, unwantedId));
428                    }
429                    ps.setString(1, filterValue);
430                    ps.setString(2, unwantedId);
431                    ps.execute();
432                }
433            } catch (SQLException e) {
434                throw new DirectoryException("failed to remove unwanted links for " + filterValue, e);
435            } finally {
436                try {
437                    if (ps != null) {
438                        ps.close();
439                    }
440                } catch (SQLException sqle) {
441                    throw new DirectoryException(sqle);
442                }
443            }
444        }
445
446        if (!idsToAdd.isEmpty()) {
447            // add missing links
448            if (filterColumn.equals(sourceColumn)) {
449                for (String missingId : idsToAdd) {
450                    addLink(filterValue, missingId, session, false);
451                }
452            } else {
453                for (String missingId : idsToAdd) {
454                    addLink(missingId, filterValue, session, false);
455                }
456            }
457        }
458    }
459
460    public void setSourceIdsForTarget(String targetId, List<String> sourceIds, SQLSession session)
461            throws DirectoryException {
462        setIdsFor(sourceColumn, sourceIds, targetColumn, targetId, session);
463    }
464
465    public void setTargetIdsForSource(String sourceId, List<String> targetIds, SQLSession session)
466            throws DirectoryException {
467        setIdsFor(targetColumn, targetIds, sourceColumn, sourceId, session);
468    }
469
470    @Override
471    public void setSourceIdsForTarget(String targetId, List<String> sourceIds) throws DirectoryException {
472        try (SQLSession session = getSQLSession()) {
473            setSourceIdsForTarget(targetId, sourceIds, session);
474        }
475    }
476
477    @Override
478    public void setSourceIdsForTarget(String targetId, List<String> sourceIds, Session session)
479            throws DirectoryException {
480        SQLSession sqlSession = (SQLSession) session;
481        maybeInitialize(sqlSession);
482        setSourceIdsForTarget(targetId, sourceIds, sqlSession);
483    }
484
485    @Override
486    public void setTargetIdsForSource(String sourceId, List<String> targetIds) throws DirectoryException {
487        try (SQLSession session = getSQLSession()) {
488            setTargetIdsForSource(sourceId, targetIds, session);
489        }
490    }
491
492    @Override
493    public void setTargetIdsForSource(String sourceId, List<String> targetIds, Session session)
494            throws DirectoryException {
495        SQLSession sqlSession = (SQLSession) session;
496        maybeInitialize(sqlSession);
497        setTargetIdsForSource(sourceId, targetIds, sqlSession);
498    }
499
500    // TODO add support for the ListDiff type
501
502    protected SQLSession getSQLSession() throws DirectoryException {
503        if (!initialized) {
504            try (SQLSession sqlSession = (SQLSession) getSourceDirectory().getSession()) {
505                initialize(sqlSession);
506                initialized = true;
507            }
508        }
509        return (SQLSession) getSourceDirectory().getSession();
510    }
511
512    /**
513     * Initialize if needed, using an existing session.
514     *
515     * @param sqlSession
516     * @throws DirectoryException
517     */
518    protected void maybeInitialize(SQLSession sqlSession) throws DirectoryException {
519        if (!initialized) {
520            initialize(sqlSession);
521            initialized = true;
522        }
523    }
524
525    public Table getTable() throws DirectoryException {
526        if (table == null) {
527            boolean nativeCase = getSQLSourceDirectory().useNativeCase();
528            table = SQLHelper.addTable(tableName, getDialect(), nativeCase);
529            SQLHelper.addColumn(table, sourceColumn, ColumnType.STRING, nativeCase);
530            SQLHelper.addColumn(table, targetColumn, ColumnType.STRING, nativeCase);
531            // index added for Azure
532            table.addIndex(null, IndexType.MAIN_NON_PRIMARY, sourceColumn);
533        }
534        return table;
535    }
536
537    private Dialect getDialect() throws DirectoryException {
538        if (dialect == null) {
539            dialect = getSQLSourceDirectory().getDialect();
540        }
541        return dialect;
542    }
543
544    public String getSourceColumn() {
545        return sourceColumn;
546    }
547
548    public String getTargetColumn() {
549        return targetColumn;
550    }
551
552    public String getTargetDirectoryName() {
553        return targetDirectoryName;
554    }
555
556    public String getTableName() {
557        return tableName;
558    }
559
560    public String getDataFileName() {
561        return dataFileName;
562    }
563
564}