001 package org.apache.ojb.broker.util.batch;
002
003 /* Copyright 2002-2005 The Apache Software Foundation
004 *
005 * Licensed under the Apache License, Version 2.0 (the "License");
006 * you may not use this file except in compliance with the License.
007 * You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017
018 import java.lang.reflect.Proxy;
019 import java.sql.Connection;
020 import java.sql.PreparedStatement;
021 import java.sql.SQLException;
022 import java.sql.Statement;
023 import java.util.ArrayList;
024 import java.util.Collection;
025 import java.util.HashMap;
026 import java.util.HashSet;
027 import java.util.Iterator;
028 import java.util.List;
029
030 import org.apache.ojb.broker.PersistenceBroker;
031 import org.apache.ojb.broker.metadata.ClassDescriptor;
032 import org.apache.ojb.broker.metadata.CollectionDescriptor;
033 import org.apache.ojb.broker.metadata.DescriptorRepository;
034 import org.apache.ojb.broker.metadata.JdbcConnectionDescriptor;
035 import org.apache.ojb.broker.metadata.ObjectReferenceDescriptor;
036 import org.apache.ojb.broker.util.WrappedConnection;
037
038 /**
039 * The implementation of {@link java.sql.Connection} which
040 * automatically gathers INSERT, UPDATE and DELETE
041 * PreparedStatements into batches.
042 *
043 * @author Oleg Nitz (<a href="mailto:olegnitz@apache.org">olegnitz@apache.org</a>)
044 * @version $Id: BatchConnection.java,v 1.1 2007-08-24 22:17:42 ewestfal Exp $
045 */
046 public class BatchConnection extends WrappedConnection
047 {
048 private static final int MAX_COUNT = 100;
049
050 /**
051 * Maps PBKey to another HashMap,
052 * which maps table name to List of related tables (N:1 or 1:1)
053 */
054 private static HashMap _pbkeyToFKInfo = new HashMap();
055
056 private boolean _useBatchInserts = true;
057 private HashMap _statements = new HashMap();
058 private ArrayList _order = new ArrayList();
059 private HashMap _fkInfo;
060 private HashSet _deleted;
061 private HashSet _dontInsert;
062 private HashSet _touched = new HashSet();
063 private int count = 0;
064 private JdbcConnectionDescriptor m_jcd;
065
066 public BatchConnection(Connection conn, PersistenceBroker broker)
067 {
068 super(conn);
069 m_jcd = broker.serviceConnectionManager().getConnectionDescriptor();
070 _fkInfo = (HashMap) _pbkeyToFKInfo.get(broker.getPBKey());
071 if (_fkInfo != null)
072 {
073 return;
074 }
075
076 DescriptorRepository repos = broker.getDescriptorRepository();
077 _fkInfo = new HashMap();
078 for (Iterator it = repos.iterator(); it.hasNext();)
079 {
080 ClassDescriptor desc = (ClassDescriptor) it.next();
081 List ordList = desc.getObjectReferenceDescriptors();
082 if (!ordList.isEmpty())
083 {
084 HashSet fkTables = getFKTablesFor(desc.getFullTableName());
085 for (Iterator it2 = ordList.iterator(); it2.hasNext();)
086 {
087 ObjectReferenceDescriptor ord = (ObjectReferenceDescriptor) it2.next();
088 ClassDescriptor oneDesc = repos.getDescriptorFor(ord.getItemClass());
089 fkTables.addAll(getFullTableNames(oneDesc, repos));
090 }
091 }
092
093 List codList = desc.getCollectionDescriptors();
094 for (Iterator it2 = codList.iterator(); it2.hasNext();)
095 {
096 CollectionDescriptor cod = (CollectionDescriptor) it2.next();
097 ClassDescriptor manyDesc = repos.getDescriptorFor(cod.getItemClass());
098 if (cod.isMtoNRelation())
099 {
100 HashSet fkTables = getFKTablesFor(cod.getIndirectionTable());
101 fkTables.addAll(getFullTableNames(desc, repos));
102 fkTables.addAll(getFullTableNames(manyDesc, repos));
103 }
104 else
105 {
106 HashSet manyTableNames = getFullTableNames(manyDesc, repos);
107 for (Iterator it3 = manyTableNames.iterator(); it3.hasNext();)
108 {
109 HashSet fkTables = getFKTablesFor((String) it3.next());
110 fkTables.addAll(getFullTableNames(desc, repos));
111 }
112 }
113 }
114 }
115 _pbkeyToFKInfo.put(broker.getPBKey(), _fkInfo);
116 }
117
118 private HashSet getFKTablesFor(String tableName)
119 {
120 HashSet fkTables = (HashSet) _fkInfo.get(tableName);
121
122 if (fkTables == null)
123 {
124 fkTables = new HashSet();
125 _fkInfo.put(tableName, fkTables);
126 }
127 return fkTables;
128 }
129
130 private HashSet getFullTableNames(ClassDescriptor desc, DescriptorRepository repos)
131 {
132 String tableName;
133 HashSet tableNamesSet = new HashSet();
134 Collection extents = desc.getExtentClasses();
135
136 tableName = desc.getFullTableName();
137 if (tableName != null)
138 {
139 tableNamesSet.add(tableName);
140 }
141 for (Iterator it = extents.iterator(); it.hasNext();)
142 {
143 Class extClass = (Class) it.next();
144 ClassDescriptor extDesc = repos.getDescriptorFor(extClass);
145 tableName = extDesc.getFullTableName();
146 if (tableName != null)
147 {
148 tableNamesSet.add(tableName);
149 }
150 }
151 return tableNamesSet;
152 }
153
154 public void setUseBatchInserts(boolean useBatchInserts)
155 {
156 _useBatchInserts = useBatchInserts;
157 }
158
159 /**
160 * Remember the order of execution
161 */
162 void nextExecuted(String sql) throws SQLException
163 {
164 count++;
165
166 if (_order.contains(sql))
167 {
168 return;
169 }
170
171 String sqlCmd = sql.substring(0, 7);
172 String rest = sql.substring(sqlCmd.equals("UPDATE ") ? 7 // "UPDATE "
173 : 12); // "INSERT INTO " or "DELETE FROM "
174 String tableName = rest.substring(0, rest.indexOf(' '));
175 HashSet fkTables = (HashSet) _fkInfo.get(tableName);
176
177 // we should not change order of INSERT/DELETE/UPDATE
178 // statements for the same table
179 if (_touched.contains(tableName))
180 {
181 executeBatch();
182 }
183 if (sqlCmd.equals("INSERT "))
184 {
185 if (_dontInsert != null && _dontInsert.contains(tableName))
186 {
187 // one of the previous INSERTs contained a table
188 // that references this table.
189 // Let's execute that previous INSERT right now so that
190 // in the future INSERTs into this table will go first
191 // in the _order array.
192 executeBatch();
193 }
194 }
195 else
196 //if (sqlCmd.equals("DELETE ") || sqlCmd.equals("UPDATE "))
197 {
198 // We process UPDATEs in the same way as DELETEs
199 // because setting FK to NULL in UPDATE is equivalent
200 // to DELETE from the referential integrity point of view.
201
202 if (_deleted != null && fkTables != null)
203 {
204 HashSet intersection = (HashSet) _deleted.clone();
205
206 intersection.retainAll(fkTables);
207 if (!intersection.isEmpty())
208 {
209 // one of the previous DELETEs contained a table
210 // that is referenced from this table.
211 // Let's execute that previous DELETE right now so that
212 // in the future DELETEs into this table will go first
213 // in the _order array.
214 executeBatch();
215 }
216 }
217 }
218
219 _order.add(sql);
220
221 _touched.add(tableName);
222 if (sqlCmd.equals("INSERT "))
223 {
224 if (fkTables != null)
225 {
226 if (_dontInsert == null)
227 {
228 _dontInsert = new HashSet();
229 }
230 _dontInsert.addAll(fkTables);
231 }
232 }
233 else if (sqlCmd.equals("DELETE "))
234 {
235 if (_deleted == null)
236 {
237 _deleted = new HashSet();
238 }
239 _deleted.add(tableName);
240 }
241 }
242
243 /**
244 * If UPDATE, INSERT or DELETE, return BatchPreparedStatement,
245 * otherwise return null.
246 */
247 private PreparedStatement prepareBatchStatement(String sql)
248 {
249 String sqlCmd = sql.substring(0, 7);
250
251 if (sqlCmd.equals("UPDATE ") || sqlCmd.equals("DELETE ") || (_useBatchInserts && sqlCmd.equals("INSERT ")))
252 {
253 PreparedStatement stmt = (PreparedStatement) _statements.get(sql);
254 if (stmt == null)
255 {
256 // [olegnitz] for JDK 1.2 we need to list both PreparedStatement and Statement
257 // interfaces, otherwise proxy.jar works incorrectly
258 stmt = (PreparedStatement) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{
259 PreparedStatement.class, Statement.class, BatchPreparedStatement.class},
260 new PreparedStatementInvocationHandler(this, sql, m_jcd));
261 _statements.put(sql, stmt);
262 }
263 return stmt;
264 }
265 else
266 {
267 return null;
268 }
269 }
270
271 public PreparedStatement prepareStatement(String sql) throws SQLException
272 {
273 PreparedStatement stmt = null;
274 stmt = prepareBatchStatement(sql);
275
276 if (stmt == null)
277 {
278 stmt = getDelegate().prepareStatement(sql);
279 }
280 return stmt;
281 }
282
283 public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)
284 throws SQLException
285 {
286 PreparedStatement stmt = null;
287 stmt = prepareBatchStatement(sql);
288
289 if (stmt == null)
290 {
291 stmt = getDelegate().prepareStatement(sql, resultSetType, resultSetConcurrency);
292 }
293 return stmt;
294 }
295
296 public void executeBatch() throws SQLException
297 {
298 BatchPreparedStatement batchStmt;
299 Connection conn = getDelegate();
300
301 try
302 {
303 for (Iterator it = _order.iterator(); it.hasNext();)
304 {
305 batchStmt = (BatchPreparedStatement) _statements.get(it.next());
306 batchStmt.doExecute(conn);
307 }
308 }
309 finally
310 {
311 _order.clear();
312
313 if (_dontInsert != null)
314 {
315 _dontInsert.clear();
316 }
317
318 if (_deleted != null)
319 {
320 _deleted.clear();
321 }
322 _touched.clear();
323 count = 0;
324 }
325 }
326
327 public void executeBatchIfNecessary() throws SQLException
328 {
329 if (count >= MAX_COUNT)
330 {
331 executeBatch();
332 }
333 }
334
335 public void clearBatch()
336 {
337 _order.clear();
338 _statements.clear();
339
340 if (_dontInsert != null)
341 {
342 _dontInsert.clear();
343 }
344
345 if (_deleted != null)
346 {
347 _deleted.clear();
348 }
349 }
350
351 public void commit() throws SQLException
352 {
353 executeBatch();
354 _statements.clear();
355 getDelegate().commit();
356 }
357
358 public void rollback() throws SQLException
359 {
360 clearBatch();
361 getDelegate().rollback();
362 }
363 }