review.fuel-infra Code Review - packages/trusty/mysql-wsrep-5.6.git/blob

   1 /*
   2    Copyright (c) 2010, 2011, Oracle and/or its affiliates. All rights reserved.
   3
   4    This program is free software; you can redistribute it and/or modify
   5    it under the terms of the GNU General Public License as published by
   6    the Free Software Foundation; version 2 of the License.
   7
   8    This program is distributed in the hope that it will be useful,
   9    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11    GNU General Public License for more details.
  12
  13    You should have received a copy of the GNU General Public License
  14    along with this program; if not, write to the Free Software
  15    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
  16 */
  17
  18 package com.mysql.clusterj.openjpa;
  19
  20 import com.mysql.clusterj.ClusterJException;
  21 import com.mysql.clusterj.ClusterJFatalInternalException;
  22 import com.mysql.clusterj.ClusterJUserException;
  23
  24 import com.mysql.clusterj.core.metadata.AbstractDomainFieldHandlerImpl;
  25 import com.mysql.clusterj.core.query.QueryDomainTypeImpl;
  26 import com.mysql.clusterj.core.spi.DomainTypeHandler;
  27 import com.mysql.clusterj.core.spi.QueryExecutionContext;
  28 import com.mysql.clusterj.core.spi.SessionSPI;
  29 import com.mysql.clusterj.core.spi.ValueHandler;
  30 import com.mysql.clusterj.core.store.IndexScanOperation;
  31 import com.mysql.clusterj.core.store.IndexScanOperation.BoundType;
  32 import com.mysql.clusterj.core.store.Dictionary;
  33 import com.mysql.clusterj.core.store.Operation;
  34 import com.mysql.clusterj.core.store.PartitionKey;
  35 import com.mysql.clusterj.core.store.ResultData;
  36 import com.mysql.clusterj.core.store.ScanFilter;
  37 import com.mysql.clusterj.core.store.Table;
  38 import com.mysql.clusterj.core.store.ScanFilter.BinaryCondition;
  39 import com.mysql.clusterj.core.util.I18NHelper;
  40 import com.mysql.clusterj.core.util.Logger;
  41 import com.mysql.clusterj.core.util.LoggerFactoryService;
  42
  43 import com.mysql.clusterj.query.Predicate;
  44 import com.mysql.clusterj.query.PredicateOperand;
  45 import com.mysql.clusterj.query.QueryDomainType;
  46
  47 import java.lang.reflect.Field;
  48 import java.sql.SQLException;
  49 import java.util.Arrays;
  50 import java.util.BitSet;
  51 import java.util.HashMap;
  52 import java.util.Map;
  53 import java.util.Set;
  54
  55 import org.apache.openjpa.jdbc.kernel.JDBCFetchConfiguration;
  56 import org.apache.openjpa.jdbc.meta.ClassMapping;
  57 import org.apache.openjpa.jdbc.meta.FieldMapping;
  58 import org.apache.openjpa.jdbc.meta.JavaSQLTypes;
  59
  60 import org.apache.openjpa.jdbc.meta.strats.RelationFieldStrategy;
  61 import org.apache.openjpa.jdbc.meta.strats.RelationStrategies;
  62 import org.apache.openjpa.jdbc.schema.Column;
  63 import org.apache.openjpa.jdbc.schema.ForeignKey;
  64 import org.apache.openjpa.jdbc.schema.Index;
  65 import org.apache.openjpa.kernel.OpenJPAStateManager;
  66 import org.apache.openjpa.meta.JavaTypes;
  67 import org.apache.openjpa.util.IntId;
  68 import org.apache.openjpa.util.LongId;
  69 import org.apache.openjpa.util.ObjectId;
  70 import org.apache.openjpa.util.OpenJPAId;
  71 import org.apache.openjpa.util.StringId;
  72
  73 /**
  74  *
  75  */
  76 public class NdbOpenJPADomainFieldHandlerImpl extends AbstractDomainFieldHandlerImpl {
  77
  78     /** My message translator */
  79     static final I18NHelper local = I18NHelper.getInstance(NdbOpenJPADomainFieldHandlerImpl.class);
  80
  81     /** My logger */
  82     static final Logger logger = LoggerFactoryService.getFactory().getInstance(NdbOpenJPADomainFieldHandlerImpl.class);
  83
  84     private static com.mysql.clusterj.core.store.Column[] emptyStoreColumns = new com.mysql.clusterj.core.store.Column[] {};
  85     /** The openjpa field mapping for this field */
  86     private FieldMapping fieldMapping;
  87     /** For single-column mappings, the mapped column */
  88     private Column column;
  89     /** The openjpa javaType of this field from org.apache.openjpa.meta.JavaTypes */
  90     private int javaType;
  91     /** The java.lang.reflect.Field in the oid class corresponding to the pk field */
  92     private Field oidField;
  93     /** The name of the class of this type, used for messages */
  94     private String javaTypeName;
  95
  96     /** True if this field is a relation */
  97     private boolean isRelation = false;
  98     /** True if this field is mapped by the relation field in the other class */
  99     private boolean isMappedBy = false;
 100     /** True if the field is mapped to a single-valued field in the other class */
 101     private boolean isToOne = false;
 102     /** True if the field is embedded */
 103     private boolean isEmbedded;
 104     /** The openjpa class mapping of the related class */
 105     private ClassMapping relatedTypeMapping = null;
 106     /** The domain type handler of the related class */
 107     private NdbOpenJPADomainTypeHandlerImpl<?> relatedDomainTypeHandler = null;
 108     /** The openjpa field mapping for the related field */
 109     private FieldMapping relatedFieldMapping;
 110     /** The domain field handler for the related field */
 111     private FieldMapping mappedByMapping;
 112     /** The class of the related type */
 113     private Class<?> relatedType = null;
 114     /** The name of the class of the related type */
 115     private String relatedTypeName = null;
 116
 117     /** These fields are to manage composite key relationships */
 118     /** The openjpa columns mapped by this relationship */
 119     private Column[] columns;
 120
 121     /** The store Columns mapped by this relationship */
 122     private com.mysql.clusterj.core.store.Column[] storeColumns = emptyStoreColumns;
 123
 124     /** The name of the related field */
 125     private String relatedFieldName;
 126
 127     /** If this field is supported for clusterjpa */
 128     private boolean supported = true;
 129
 130     /** The reason the field is not supported */
 131     private String reason = "";
 132
 133     private RelatedFieldLoadManager relatedFieldLoadManager;
 134
 135     public FieldMapping getFieldMapping() {
 136         return fieldMapping;
 137     }
 138
 139     public NdbOpenJPADomainFieldHandlerImpl(Dictionary dictionary, NdbOpenJPADomainTypeHandlerImpl<?> domainTypeHandler,
 140             NdbOpenJPAConfigurationImpl domainTypeHandlerFactory, final FieldMapping fieldMapping) {
 141
 142         String message = null;
 143         this.fieldMapping = fieldMapping;
 144         this.domainTypeHandler = domainTypeHandler;
 145         this.name = fieldMapping.getName();
 146         this.fieldNumber = fieldMapping.getIndex();
 147         this.primaryKey = fieldMapping.isPrimaryKey();
 148         this.columns = fieldMapping.getColumns();
 149         this.objectOperationHandlerDelegate = objectOperationHandlerUnsupportedType;
 150         this.mappedByMapping = fieldMapping.getMappedByMapping();
 151         this.isMappedBy = mappedByMapping != null;
 152         this.isToOne = fieldMapping.getStrategy() instanceof RelationFieldStrategy;
 153         if (isMappedBy) {
 154             relatedType = mappedByMapping.getDeclaringType();
 155             relatedFieldMapping = fieldMapping.getMappedByMapping();
 156         }
 157         // TODO are these valid for every field?
 158         this.relatedTypeMapping = fieldMapping.getDeclaredTypeMapping();
 159         if (relatedTypeMapping != null) {
 160             relatedType = relatedTypeMapping.getDescribedType();
 161         }
 162         if (relatedType != null) {
 163             relatedTypeName = relatedType.getName();
 164         }
 165         // TODO: the following might not be definitive to identify a relationship
 166         this.isRelation = fieldMapping.getStrategy().getClass().getName().contains("Relation");
 167         this.isEmbedded = fieldMapping.getStrategy().getClass().getName().contains("Embed");
 168         if (logger.isDetailEnabled()) logger.detail(
 169                 "field: " + name + " strategy: " + fieldMapping.getStrategy().getClass().getName() + " with " + columns.length + " columns.");
 170         if ((!(isRelation | isEmbedded))) {
 171             if (columns.length == 1) {
 172                 // each field is mapped to one column
 173                 this.column = columns[0];
 174                 this.columnName = column.getName();
 175                 Table table = domainTypeHandler.getTable();
 176                 if (table == null) {
 177                     message = local.message("ERR_No_Mapped_Table", domainTypeHandler.getName());
 178                     setUnsupported(message);
 179                     return;
 180                 }
 181                 this.storeColumn = table.getColumn(columnName);
 182                 if (storeColumn == null) {
 183                     message = local.message("ERR_No_Column", name, table.getName(), columnName);
 184                     setUnsupported(message);
 185                     return;
 186                 }
 187                 this.storeColumns = new com.mysql.clusterj.core.store.Column[] {storeColumn};
 188                 charsetName = storeColumn.getCharsetName();
 189                 // set up the default object operation handler for the column type
 190                 // TODO this might better use the "Class type;" field in superclass
 191                 this.javaType = column.getJavaType();
 192                 this.objectOperationHandlerDelegate = getObjectOperationHandler(javaType);
 193                 if (objectOperationHandlerUnsupportedType.equals(objectOperationHandlerDelegate)) {
 194                     message = local.message("ERR_Unsupported_Meta_Type", javaType);
 195                     setUnsupported(message);
 196                     return;
 197                 } else {
 198                     this.javaTypeName = NdbOpenJPAUtility.getJavaTypeName(javaType);
 199                     if (storeColumn.isPrimaryKey()) {
 200                         domainTypeHandler.registerPrimaryKeyColumn(this, storeColumn.getName());
 201                     }
 202                 }
 203             } else if (columns.length > 1) {
 204                 // error, no support
 205                 StringBuffer buffer = new StringBuffer();
 206                 String separator = "";
 207                 for (Column errorColumn : columns) {
 208                     buffer.append(separator);
 209                     buffer.append(errorColumn.getName());
 210                     separator = ", ";
 211                 }
 212                 message = local.message("ERR_More_Than_One_Column_Mapped_To_A_Field",
 213                         domainTypeHandler.getName(), name, buffer);
 214                 logger.info(message);
 215                 setUnsupported(message);
 216                 return;
 217             } else if (columns.length == 0) {
 218                 message = local.message("ERR_No_Column_Mapped_To_A_Field",
 219                         domainTypeHandler.getName(), name, fieldMapping.getTable(), fieldMapping.getStrategy().getClass().getName());
 220                 logger.info(message);
 221                 setUnsupported(message);
 222                 return;
 223             }
 224             if (this.primaryKey) {
 225                 // each field is mapped to its own column
 226                 // if using a user-defined openJPAId class, set up the value handler
 227                 oidField = getFieldForOidClass(this, domainTypeHandler.getOidClass(), name);
 228                 indexNames.add("PRIMARY");
 229                 switch (javaType) {
 230                     case JavaTypes.INT:
 231                         this.objectOperationHandlerDelegate = objectOperationHandlerKeyInt;
 232                         break;
 233                     case JavaTypes.INT_OBJ:
 234                         this.objectOperationHandlerDelegate = objectOperationHandlerKeyObjectInteger;
 235                         break;
 236                     case JavaTypes.LONG:
 237                         this.objectOperationHandlerDelegate = objectOperationHandlerKeyLong;
 238                         break;
 239                     case JavaTypes.LONG_OBJ:
 240                         this.objectOperationHandlerDelegate = objectOperationHandlerKeyObjectLong;
 241                         break;
 242                    case JavaTypes.STRING: this.objectOperationHandlerDelegate =
 243                         objectOperationHandlerKeyString;
 244                         break;
 245                     default:
 246                         message = local.message("ERR_Illegal_Primary_Key_Type",
 247                             domainTypeHandler.getName(), name, columnName, javaTypeName);
 248                         logger.info(message);
 249                         setUnsupported(message);
 250                 }
 251             }
 252         } else if (isRelation) {
 253             // relationships might have zero, one, or more columns
 254             if (columns.length == 1) {
 255                 this.column = columns[0];
 256                 this.columnName = column.getName();
 257                 this.columnNames = new String[] {columnName};
 258                 Table table = domainTypeHandler.getTable();
 259                 this.storeColumn = table.getColumn(columnName);
 260                 if (storeColumn == null) {
 261                     message = local.message("ERR_No_Column", name, table.getName(), columnName);
 262                     setUnsupported(message);
 263                     return;
 264                 }
 265                 this.storeColumns = new com.mysql.clusterj.core.store.Column[] {storeColumn};
 266                 // set up the default object operation handler for the column type
 267                 this.javaType = column.getJavaType();
 268                 this.javaTypeName = NdbOpenJPAUtility.getJavaTypeName(javaType);
 269                 this.objectOperationHandlerDelegate = getObjectOperationHandlerRelationDelegate(javaType);
 270                 if (objectOperationHandlerDelegate == null) {
 271                     // unsupported primary key type
 272                     return;
 273                 }
 274             } else if (columns.length == 0) {
 275                 if (isMappedBy) {
 276                     // this is the case of a OneToMany field mapped by columns in another table
 277                     this.objectOperationHandlerDelegate = objectOperationHandlerVirtualType;
 278                 } else {
 279                     message = local.message("ERR_No_Columns_And_Not_Mapped_By",
 280                             this.domainTypeHandler.getName(), this.name);
 281                     logger.info(message);
 282                     setUnsupported(message);
 283                 }
 284             } else {
 285                 // multiple columns for related object
 286                 if (isMappedBy) {
 287                     // this is the case of OneToOne field mapped by columns in another table
 288                     this.objectOperationHandlerDelegate = objectOperationHandlerVirtualType;
 289                 } else {
 290                     // create an array of NdbOpenJPADomainFieldHandlerImpl
 291                     // one for each column in the foreign key
 292                     // each one needs to be able to extract the foreign key
 293                     // value from the openJPAId instance of the related instance
 294                     // using the oidField object
 295                     this.relatedTypeMapping = fieldMapping.getDeclaredTypeMapping();
 296                     this.relatedType = relatedTypeMapping.getDescribedType();
 297                     Class<?> oid = relatedTypeMapping.getObjectIdType();
 298                     if (logger.isDetailEnabled()) logger.detail(
 299                             "For class: " + domainTypeHandler.getName() +
 300                             " field: " + name + " related type is: " + relatedType.getName() +
 301                             " objectid type: " + oid.getName());
 302                     // create the domain field handlers for each column
 303                     this.compositeDomainFieldHandlers = new NdbOpenJPADomainFieldHandlerImpl[columns.length];
 304                     this.columnNames = new String[columns.length];
 305                     this.storeColumns = new com.mysql.clusterj.core.store.Column[columns.length];
 306                     for (int i = 0; i < columns.length; ++i) {
 307                         StringBuffer detailMessage = new StringBuffer();
 308                         Column localColumn = columns[i];
 309                         String localColumnName = localColumn.getName();
 310                         Table table = domainTypeHandler.getTable();
 311                         com.mysql.clusterj.core.store.Column localStoreColumn = table.getColumn(localColumnName);
 312                         if (localStoreColumn == null) {
 313                             message = local.message("ERR_No_Column", name, table.getName(), localColumnName);
 314                             logger.info(message);
 315                             setUnsupported(message);
 316                             return;
 317                         }
 318                         this.storeColumns[i] = localStoreColumn;
 319                         this.columnNames[i] = localColumnName;
 320                         ForeignKey foreignKey = fieldMapping.getForeignKey();
 321                         // get the primary key column corresponding to the local column
 322                         Column pkColumn = foreignKey.getPrimaryKeyColumn(localColumn);
 323                         if (logger.isDetailEnabled()) {
 324                             detailMessage.append(" column: " + localColumnName);
 325                             detailMessage.append(" fk-> " + foreignKey);
 326                             detailMessage.append(" pkColumn-> " + pkColumn);
 327                             logger.detail(detailMessage.toString());
 328                         }
 329                         NdbOpenJPADomainFieldHandlerImpl relatedFieldHandler =
 330                             new NdbOpenJPADomainFieldHandlerImpl(this, localColumn, pkColumn);
 331                         if (relatedFieldHandler.isSupported()) {
 332                             this.compositeDomainFieldHandlers[i] = relatedFieldHandler;
 333                         } else {
 334                             message = relatedFieldHandler.getReason();
 335                             setUnsupported(message);
 336                             return;
 337                         }
 338
 339                     }
 340                     this.objectOperationHandlerDelegate =
 341                             objectOperationHandlerRelationCompositeField;
 342                 }
 343             }
 344         } else {
 345             // embedded field
 346             message = local.message("ERR_Embedded_Fields_Not_Supported",
 347                     this.domainTypeHandler.getName(), this.name);
 348             logger.info(message);
 349             setUnsupported(message);
 350             return;
 351         }
 352         // now handle indexes, for supported field types
 353         Index index = fieldMapping.getJoinIndex();
 354         // TODO: where is this annotation used?
 355         if (index != null) {
 356             String indexName = index.getName();
 357             Column[] indexColumns = index.getColumns();
 358             if (logger.isDetailEnabled()) {
 359                 StringBuilder buffer = new StringBuilder("Found index name ");
 360                 buffer.append(indexName);
 361                 buffer.append(" [");
 362                 for (Column indexColumn : indexColumns) {
 363                     if (logger.isDetailEnabled()) {
 364                         buffer.append(indexColumn.getName());
 365                         buffer.append(" ");
 366                     }
 367                 }
 368                 buffer.append("]");
 369                 logger.detail(buffer.toString());
 370             }
 371         }
 372         index = fieldMapping.getValueIndex();
 373         // Value indexes are used for ManyToOne and OneToOne relationship indexes on the mapped side
 374         if (index != null) {
 375             StringBuffer buffer = null;
 376             if (logger.isDetailEnabled())  buffer = new StringBuffer("Found index ");
 377             String indexName = index.getName();
 378             if (logger.isDetailEnabled())  buffer.append(indexName + " [ ");
 379             Column[] indexColumns = index.getColumns();
 380             for (Column indexColumn : indexColumns) {
 381                 if (logger.isDetailEnabled()) buffer.append(indexColumn.getName() + " ");
 382             }
 383             if (logger.isDetailEnabled()) buffer.append("]");
 384             if (logger.isDetailEnabled()) logger.detail(buffer.toString());
 385             // create an index entry for clusterj queries
 386             // Create an index handler and register this instance with the domain type handler
 387             indices = domainTypeHandler.createIndexHandler(this, dictionary, indexName);
 388         }
 389         this.type = fieldMapping.getType();
 390         if (logger.isTraceEnabled()) {
 391             logger.trace(
 392                     " number: " + this.fieldNumber +
 393                     " name: " + this.name +
 394                     " column: " + this.columnName +
 395                     " Java type: " + javaType +
 396                     " strategy: " + toString(fieldMapping.getStrategy()) +
 397                     " ObjectOperationHandler: " + objectOperationHandlerDelegate.handler());
 398         }
 399     }
 400
 401     /** Initialize relationship handling. There are three types of relationships
 402      * supported by clusterjpa:
 403      * <ul><li>direct ToOne relationship mapped by a foreign key on this side
 404      * </li><li>ToOne relationship mapped by a foreign key on the other side
 405      * </li><li>ToMany relationship mapped by a foreign key on the other side
 406      * </li></ul>
 407      *
 408      */
 409     public void initializeRelations() {
 410         if (isRelation) {
 411             // set up related field load handler
 412             if (isMappedBy && isToOne) {
 413                 // mapped by other side with one instance
 414                 this.relatedDomainTypeHandler = ((NdbOpenJPADomainTypeHandlerImpl<?>)domainTypeHandler)
 415                         .registerDependency(relatedTypeMapping);
 416                 this.relatedFieldName = relatedFieldMapping.getName();
 417                 relatedFieldLoadManager = new RelatedFieldLoadManager() {
 418                     public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store,
 419                             JDBCFetchConfiguration fetch) throws SQLException {
 420                         SessionSPI session = store.getSession();
 421                         session.startAutoTransaction();
 422                         NdbOpenJPAResult queryResult = queryRelated(sm, store);
 423                         Object related = null;
 424                         try {
 425                             if (queryResult.next()) {
 426                                 // instantiate the related object from the result of the query
 427                                 related = store.load(relatedTypeMapping, fetch, (BitSet) null, queryResult);
 428                             }
 429                             if (logger.isDetailEnabled()) logger.detail("related object is: " + related);
 430                             // store the value of the related object in this field
 431                             sm.storeObjectField(fieldNumber, related);
 432                             session.endAutoTransaction();
 433                         } catch (Exception e) {
 434                             session.failAutoTransaction();
 435                         }
 436                     }
 437                 };
 438                 if (logger.isDetailEnabled()) logger.detail("Single-valued relationship field " + name
 439                         + " is mapped by " + relatedTypeName + " field " + relatedFieldName
 440                         + " with relatedDomainTypeHandler " + relatedDomainTypeHandler.getName());
 441             } else if (isMappedBy && !isToOne) {
 442                 // mapped by other side with multiple instances
 443                 this.relatedTypeMapping = mappedByMapping.getDeclaringMapping();
 444                 this.relatedDomainTypeHandler = ((NdbOpenJPADomainTypeHandlerImpl<?>)domainTypeHandler)
 445                         .registerDependency(relatedTypeMapping);
 446                 this.relatedFieldName = mappedByMapping.getName();
 447                 relatedTypeName = relatedDomainTypeHandler.getName();
 448                 if (logger.isDetailEnabled()) logger.detail("Multi-valued relationship field " + name
 449                         + " is mapped by " + relatedTypeName + " field " + relatedFieldName);
 450                     relatedFieldLoadManager = new RelatedFieldLoadManager() {
 451                         public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store,
 452                                 JDBCFetchConfiguration fetch) throws SQLException {
 453                             SessionSPI session = store.getSession();
 454                             session.startAutoTransaction();
 455                             try {
 456                                 NdbOpenJPAResult queryResult = queryRelated(sm, store);
 457                                 while (queryResult.next()) {
 458                                     if (logger.isDetailEnabled()) logger.detail("loading related instance of type: " + relatedTypeMapping.getDescribedType().getName());
 459                                     store.load(relatedTypeMapping, fetch, (BitSet) null, queryResult);
 460                                 }
 461                                 fieldMapping.load(sm, store, fetch);
 462                                 session.endAutoTransaction();
 463                             } catch (Exception e) {
 464                                 session.failAutoTransaction();
 465                                 throw new ClusterJException(local.message("ERR_Exception_While_Loading"), e);
 466                             }
 467                     }
 468                 };
 469             } else {
 470                 // this side contains foreign key to other side
 471                 if (logger.isDetailEnabled()) logger.detail("NdbOpenJPADomainFieldHandlerImpl.initializeRelations for "
 472                         + fieldMapping.getName() + " column " + (column==null?"null":column.getName())
 473                         + " relatedFieldName " + relatedFieldName
 474                         + " relatedFieldMapping " + relatedFieldMapping
 475                         + " relatedTypeMapping " + relatedTypeMapping);
 476                 // record dependency to related type if not null
 477                 if (relatedTypeMapping != null) {
 478                     this.relatedDomainTypeHandler = ((NdbOpenJPADomainTypeHandlerImpl<?>)domainTypeHandler)
 479                         .registerDependency(relatedTypeMapping);
 480                     relatedFieldLoadManager = new RelatedFieldLoadManager() {
 481                         public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store,
 482                                 JDBCFetchConfiguration fetch) throws SQLException {
 483                             if (logger.isDetailEnabled()) logger.detail("Loading field " + name + "from stored key");
 484                             fieldMapping.load(sm, store, fetch);
 485                         }
 486                     };
 487                 }
 488             }
 489         }
 490     }
 491
 492     /** Get the object operation handler for the specific java field type.
 493      * @param javaType the java type from JavaTypes or JavaSQLTypes
 494      * @return the object operation handler
 495      */
 496     private ObjectOperationHandler getObjectOperationHandler(int javaType) {
 497         // the default is unsupported
 498         ObjectOperationHandler result = objectOperationHandlerUnsupportedType;
 499         // if a known java type, the default is from the table of object operation handlers
 500         if (javaType < objectOperationHandlers.length) {
 501             result = objectOperationHandlers[javaType];
 502         }
 503         // handle exceptions, including all JavaSQLTypes and special handling for String
 504         switch (javaType) {
 505             case JavaSQLTypes.SQL_DATE:
 506                 return objectOperationHandlerJavaSqlDate;
 507             case JavaSQLTypes.TIME:
 508                 return objectOperationHandlerJavaSqlTime;
 509             case JavaSQLTypes.TIMESTAMP:
 510                 return objectOperationHandlerJavaSqlTimestamp;
 511             case JavaSQLTypes.BYTES:
 512                 switch(storeColumn.getType()) {
 513                     case Blob:
 514                     case Longvarbinary:
 515                         return objectOperationHandlerBytesLob;
 516                     case Binary:
 517                     case Varbinary:
 518                         return objectOperationHandlerBytes;
 519                     default:
 520                 }
 521             case JavaTypes.STRING:
 522                 switch(storeColumn.getType()) {
 523                     case Text:
 524                         return objectOperationHandlerStringLob;
 525                     case Char:
 526                     case Varchar:
 527                     case Longvarchar:
 528                         return objectOperationHandlerString;
 529                     default:
 530                 }
 531             default:
 532         }
 533         return result;
 534     }
 535
 536     /** This field handler is used with compound "foreign keys". Each column of
 537      * the compound "foreign key" has its own field handler. The parent
 538      * field handler has the relationship field but no columns.
 539      *
 540      * @param parent the field handler with the relationship field
 541      * @param localColumn the "foreign key" column in this table
 542      * @param pkColumn the primary key column in the other table
 543      */
 544     public NdbOpenJPADomainFieldHandlerImpl(NdbOpenJPADomainFieldHandlerImpl parent,
 545             Column localColumn, Column pkColumn) {
 546         String message = null;
 547         if (logger.isDetailEnabled()) logger.detail("NdbOpenJPADomainFieldHandlerImpl<init> for localColumn: " + localColumn + " pkColumn: " + pkColumn);
 548         this.column = localColumn;
 549         Table table = parent.domainTypeHandler.getStoreTable();
 550         this.storeColumn = table.getColumn(localColumn.getName());
 551         if (storeColumn == null) {
 552             message = local.message("ERR_No_Column", parent.getName(), table.getName(), columnName);
 553             setUnsupported(message);
 554             logger.info(message);
 555             return;
 556         }
 557         this.javaType = column.getJavaType();
 558         this.objectOperationHandlerDelegate = getObjectOperationHandlerRelationDelegate(javaType);
 559         if (objectOperationHandlerDelegate == null) {
 560             // unsupported primary key type
 561             return;
 562         }
 563         this.columnName = column.getName();
 564         this.fieldNumber = parent.fieldNumber;
 565         this.domainTypeHandler = parent.domainTypeHandler;
 566         this.relatedTypeMapping = parent.relatedTypeMapping;
 567         if (relatedTypeMapping != null) {
 568             relatedType = relatedTypeMapping.getDescribedType();
 569             if (relatedType != null) {
 570                 relatedTypeName = relatedType.getName();
 571             }
 572         }
 573         // now find the field in the related class corresponding to this column
 574         FieldMapping[] relatedFieldMappings = relatedTypeMapping.getPrimaryKeyFieldMappings();
 575         for (FieldMapping rfm: relatedFieldMappings) {
 576             Column[] rcs = rfm.getColumns();
 577             if (logger.isDetailEnabled()) logger.detail("NdbOpenJPADomainFieldHandlerImpl<init> trying primary key column: " + rcs[0]);
 578             if (rcs.length == 1 && rcs[0].equals(pkColumn)) {
 579                 // found the corresponding pk field
 580                 String pkFieldName = rfm.getName();
 581                 oidField = getFieldForOidClass(this, relatedTypeMapping.getObjectIdType(), pkFieldName);
 582             if (logger.isDetailEnabled()) logger.detail("NdbOpenJPADomainFieldHandlerImpl<init> found primary key column: " + rcs[0] + " for field: " + pkFieldName);
 583                 break;
 584             }
 585         }
 586         if (oidField == null) {
 587             message = local.message("ERR_No_Oid_Field", pkColumn);
 588             setUnsupported(message);
 589             logger.info(message);
 590             return;
 591         }
 592         if (logger.isTraceEnabled()) {
 593             logger.trace(" Relation Field Handler for column: " + columnName +
 594                     " number: " + this.fieldNumber +
 595                     " name: " + this.name +
 596                     " column: " + this.columnName +
 597                     " Java type: " + javaType +
 598                     " ObjectOperationHandler: " + objectOperationHandlerDelegate.handler());
 599         }
 600
 601     }
 602
 603     interface RelatedFieldLoadManager {
 604         public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store,
 605                 JDBCFetchConfiguration fetch) throws SQLException ;
 606     }
 607
 608     /** Load the value of this field. This will be done here for relationship
 609      * since basic fields are loaded when the instance is first initialized.
 610      *
 611      * @param sm the openjpa state manager for the instance
 612      * @param store the store manager
 613      * @param fetch the fetch configuration, presently unused
 614      * @throws SQLException
 615      */
 616     public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store, JDBCFetchConfiguration fetch)
 617             throws SQLException {
 618         if (isRelation) {
 619             relatedFieldLoadManager.load(sm, store, fetch);
 620         } else {
 621             throw new ClusterJFatalInternalException("load called for non-relationship field "
 622                     + this.getName() + " mapped to column " + this.columnName);
 623         }
 624     }
 625
 626     /** Query the related type for instance(s) whose related field refers to this instance.
 627      * @param sm the state manager
 628      * @param store the store manager
 629      * @return the result of executing a query for the related type based on this instance's primary key
 630      */
 631     private NdbOpenJPAResult queryRelated(OpenJPAStateManager sm, NdbOpenJPAStoreManager store) {
 632         // get the Oid object for this sm
 633         OpenJPAId openJPAId = (OpenJPAId)sm.getObjectId();
 634         Object thisOid = openJPAId.getIdObject();
 635         QueryDomainType<?> queryDomainType = store.createQueryDomainType(relatedType);
 636         if (logger.isDetailEnabled()) logger.detail("created query for " + queryDomainType.getType().getName());
 637         // query for related type equals this pk oid value
 638         Predicate predicate = queryDomainType.get(relatedFieldName).equal(queryDomainType.param(relatedFieldName));
 639         queryDomainType.where(predicate);
 640         Map<String, Object> parameterList = new HashMap<String, Object>();
 641         parameterList.put(relatedFieldName, thisOid);
 642         if (logger.isDetailEnabled()) logger.detail(parameterList.toString());
 643         NdbOpenJPAResult queryResult = store.executeQuery(relatedDomainTypeHandler, queryDomainType, parameterList);
 644         // debug query result
 645         if (logger.isDetailEnabled()) {
 646             DomainTypeHandler<?> handler = queryResult.domainTypeHandler;
 647             Set<String> columnNames = queryResult.getColumnNames();
 648             StringBuffer buffer = new StringBuffer("Executed query for ");
 649             buffer.append(handler.getName());
 650             buffer.append(" returned columns: ");
 651             buffer.append(Arrays.toString(columnNames.toArray()));
 652             logger.detail(buffer.toString());
 653         }
 654         return queryResult;
 655     }
 656
 657     public void load(OpenJPAStateManager sm, NdbOpenJPAStoreManager store, JDBCFetchConfiguration fetch,
 658             NdbOpenJPAResult result) throws SQLException {
 659         fieldMapping.load(sm, store, fetch, result);
 660     }
 661
 662     /** Add filters to the query and return the values to be used for the filter.
 663      *
 664      * @param queryDomainType the QueryDomainType
 665      * @param thisOid the object id to be used to query foreign keys
 666      * @return the parameter map for the query with bound data values
 667      */
 668     public Map<String, Object> createParameterMap(QueryDomainType<?> queryDomainType, Object thisOid) {
 669         return ((ObjectOperationHandlerRelationField)objectOperationHandlerDelegate).createParameterMap(this, queryDomainType, thisOid);
 670     }
 671
 672     public ObjectOperationHandler[] objectOperationHandlers =
 673             new ObjectOperationHandler[] {
 674         objectOperationHandlerBoolean,         /* 0: boolean */
 675         objectOperationHandlerByte,            /* 1: byte */
 676         objectOperationHandlerUnsupportedType, /* 2: char */
 677         objectOperationHandlerDouble,          /* 3: double */
 678         objectOperationHandlerFloat,           /* 4: float */
 679         objectOperationHandlerInt,             /* 5: int */
 680         objectOperationHandlerLong,            /* 6: long */
 681         objectOperationHandlerShort,           /* 7: short */
 682         objectOperationHandlerUnsupportedType, /* 8: Object */
 683         objectOperationHandlerString,          /* 9: String */
 684         objectOperationHandlerUnsupportedType, /* 10: Number */
 685         objectOperationHandlerUnsupportedType, /* 11: Array */
 686         objectOperationHandlerUnsupportedType, /* 12: Collection */
 687         objectOperationHandlerUnsupportedType, /* 13: Map */
 688         objectOperationHandlerJavaUtilDate,    /* 14: java.util.Date */
 689         objectOperationHandlerUnsupportedType, /* 15: PC */
 690         objectOperationHandlerObjectBoolean,   /* 16: Boolean */
 691         objectOperationHandlerObjectByte,      /* 17: Byte */
 692         objectOperationHandlerUnsupportedType, /* 18: Character */
 693         objectOperationHandlerObjectDouble,    /* 19: Double */
 694         objectOperationHandlerObjectFloat,     /* 20: Float */
 695         objectOperationHandlerObjectInteger,   /* 21: Integer */
 696         objectOperationHandlerObjectLong,      /* 22: Long */
 697         objectOperationHandlerObjectShort,     /* 23: Short */
 698         objectOperationHandlerDecimal,         /* 24: BigDecimal */
 699         objectOperationHandlerBigInteger,      /* 25: BigInteger */
 700         objectOperationHandlerUnsupportedType, /* 26: Locale */
 701         objectOperationHandlerUnsupportedType, /* 27: PC Untyped */
 702         objectOperationHandlerUnsupportedType, /* 28: Calendar */
 703         objectOperationHandlerUnsupportedType, /* 29: OID */
 704         objectOperationHandlerUnsupportedType, /* 30: InputStream */
 705         objectOperationHandlerUnsupportedType  /* 31: InputReader */
 706     };
 707
 708     public int compareTo(Object o) {
 709         return compareTo((NdbOpenJPADomainFieldHandlerImpl)o);
 710     }
 711
 712     protected String toString(Object o) {
 713         return o.getClass().getSimpleName();
 714     }
 715
 716     Column[] getColumns() {
 717         return fieldMapping.getColumns();
 718     }
 719
 720     protected Object getKeyValue(Object keys) {
 721         Object key = keys;
 722         if (keys instanceof ObjectId) {
 723             key = ((ObjectId)keys).getId();
 724         }
 725         return getKeyValue(oidField, key);
 726     }
 727
 728     protected static Object getKeyValue(Field field, Object keys) {
 729         try {
 730             Object result;
 731             String fieldName = "none";
 732             if (field != null) {
 733                 result = field.get(keys);
 734                 fieldName = field.getName();
 735             } else {
 736                 result = keys;
 737             }
 738             if (logger.isDetailEnabled()) logger.detail("For field " + fieldName + " keys: " + keys + " value returned is " + result);
 739             return result;
 740         } catch (IllegalArgumentException ex) {
 741             String message = "IllegalArgumentException, field " + field.getDeclaringClass().getName() + ":" + field.getName() + " keys: " + keys;
 742             logger.error(message);
 743             throw new ClusterJUserException(message, ex);
 744         } catch (IllegalAccessException ex) {
 745             String message = "IllegalAccessException, field " + field.getDeclaringClass().getName() + ":" + field.getName() + " keys: " + keys;
 746             throw new ClusterJUserException(message, ex);
 747         }
 748     }
 749
 750     public Field getOidField() {
 751         return oidField;
 752     }
 753
 754     protected static Field getFieldForOidClass(
 755             NdbOpenJPADomainFieldHandlerImpl ndbOpenJPADomainFieldHandlerImpl,
 756             Class<?> oidClass, String fieldName) {
 757         String message = null;
 758         Field result = null;
 759         if (logger.isDetailEnabled()) logger.detail("Oid class: " + oidClass.getName());
 760         // the openJPAId class might be a simple type or a user-defined type
 761         if (OpenJPAId.class.isAssignableFrom(oidClass)) {
 762             return null;
 763         } else {
 764             try {
 765                 // user-defined class; get Field to extract values at runtime
 766                 result = oidClass.getField(fieldName);
 767                 if (logger.isDetailEnabled()) logger.detail("OidField: " + result);
 768                 return result;
 769             } catch (NoSuchFieldException ex) {
 770                 message = local.message("ERR_No_Field_In_Oid_Class", oidClass.getName(), fieldName);
 771                 logger.info(message);
 772                 ndbOpenJPADomainFieldHandlerImpl.setUnsupported(message);
 773                 return null;
 774             } catch (SecurityException ex) {
 775                 message = local.message("ERR_Security_Violation_For_Oid_Class", oidClass.getName());
 776                 logger.info(message);
 777                 ndbOpenJPADomainFieldHandlerImpl.setUnsupported(message);
 778                 return null;
 779             }
 780         }
 781     }
 782
 783     protected ObjectOperationHandler getObjectOperationHandlerRelationDelegate(int javaType) {
 784         String message;
 785         switch (javaType) {
 786             case JavaTypes.INT:
 787             case JavaTypes.INT_OBJ:
 788                 return objectOperationHandlerRelationIntField;
 789
 790             case JavaTypes.LONG:
 791             case JavaTypes.LONG_OBJ:
 792                 return objectOperationHandlerRelationLongField;
 793
 794             case JavaTypes.STRING:
 795                 return objectOperationHandlerRelationStringField;
 796
 797             default:
 798                 message = local.message("ERR_Illegal_Foreign_Key_Type",
 799                         domainTypeHandler.getName(), name, columnName, javaType);
 800                 setUnsupported(message);
 801                 return null;
 802         }
 803     }
 804
 805     static abstract class ObjectOperationHandlerRelationField implements ObjectOperationHandler {
 806
 807         public boolean isPrimitive() {
 808             return false;
 809         }
 810
 811         /** Add the filter to the query and create the parameter map with the bound value of the oid.
 812          *
 813          * @param domainFieldHandler the domain field handler
 814          * @param queryDomainObject the query domain object
 815          * @param oid the object id value to bind to the filter
 816          * @return the map with the bound parameter
 817          */
 818         public Map<String, Object> createParameterMap(NdbOpenJPADomainFieldHandlerImpl domainFieldHandler, QueryDomainType<?> queryDomainObject, Object oid) {
 819             String name = domainFieldHandler.name;
 820             PredicateOperand parameter = queryDomainObject.get(name);
 821             Predicate predicate = parameter.equal(parameter);
 822             queryDomainObject.where(predicate);
 823             // construct a map of parameter binding to the value in oid
 824             Map<String, Object> result = new HashMap<String, Object>();
 825             Object value = domainFieldHandler.getKeyValue(oid);
 826             result.put(name, value);
 827             if (logger.isDetailEnabled()) logger.detail("Map.Entry key: " + name + ", value: " + value);
 828             return result;
 829         }
 830
 831         public void objectInitializeJavaDefaultValue(AbstractDomainFieldHandlerImpl fmd, ValueHandler handler) {
 832             throw new ClusterJFatalInternalException(
 833                     local.message("ERR_Implementation_Should_Not_Occur"));
 834         }
 835
 836         public void operationGetValue(AbstractDomainFieldHandlerImpl fmd, Operation op) {
 837             op.getValue(fmd.getStoreColumn());
 838         }
 839
 840         public Object getDefaultValueFor(AbstractDomainFieldHandlerImpl fmd, String columnDefaultValue) {
 841             throw new ClusterJFatalInternalException(
 842                     local.message("ERR_Implementation_Should_Not_Occur"));
 843         }
 844
 845         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
 846             throw new ClusterJFatalInternalException(
 847                     local.message("ERR_Implementation_Should_Not_Occur"));
 848         }
 849
 850         public void objectSetValue(AbstractDomainFieldHandlerImpl fmd, ResultData rs, ValueHandler handler) {
 851             throw new ClusterJFatalInternalException(
 852                     local.message("ERR_Implementation_Should_Not_Occur"));
 853         }
 854
 855         public void operationSetBounds(AbstractDomainFieldHandlerImpl fmd, Object value, BoundType type, IndexScanOperation op) {
 856             throw new ClusterJFatalInternalException(
 857                     local.message("ERR_Implementation_Should_Not_Occur"));
 858         }
 859
 860         public void filterCompareValue(AbstractDomainFieldHandlerImpl fmd, Object value, BinaryCondition condition, ScanFilter filter) {
 861             throw new ClusterJFatalInternalException(
 862                     local.message("ERR_Implementation_Should_Not_Occur"));
 863         }
 864
 865         public void operationEqual(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
 866             throw new ClusterJFatalInternalException(
 867                     local.message("ERR_Implementation_Should_Not_Occur"));
 868         }
 869
 870         public boolean isValidIndexType(AbstractDomainFieldHandlerImpl fmd, boolean hashNotOrdered) {
 871             // relationships can use either hash or btree indexes
 872             return true;
 873         }
 874
 875         protected int getInt(Object objectId) {
 876             // get an int value from an objectId
 877             if (objectId instanceof IntId) {
 878                 return ((IntId)objectId).getId();
 879             } else if (objectId instanceof OpenJPAId) {
 880                 OpenJPAId openJPAId = (OpenJPAId)objectId;
 881                 Object id = openJPAId.getIdObject();
 882                 if (id instanceof Integer) {
 883                     return ((Integer)id).intValue();
 884                 }
 885                 throw new UnsupportedOperationException(
 886                         local.message("ERR_Unsupported_Object_Id_Type", "int key", "OpenJPAId"));
 887             } else {
 888                 String message = (objectId == null)?"<null>":objectId.getClass().getName();
 889                 throw new UnsupportedOperationException(
 890                         local.message("ERR_Unsupported_Object_Id_Type", "int key", message));
 891             }
 892         }
 893
 894         protected long getLong(Object objectId) {
 895             // get a long value from an objectId
 896             if (objectId instanceof LongId) {
 897                 return ((LongId)objectId).getId();
 898             } else if (objectId instanceof OpenJPAId) {
 899                 OpenJPAId openJPAId = (OpenJPAId)objectId;
 900                 Object id = openJPAId.getIdObject();
 901                 if (id instanceof Long) {
 902                     return ((Long)id).longValue();
 903                 }
 904                 throw new UnsupportedOperationException(
 905                         local.message("ERR_Unsupported_Object_Id_Type", "long key", "OpenJPAId"));
 906             } else {
 907                 String message = (objectId == null)?"<null>":objectId.getClass().getName();
 908                 throw new UnsupportedOperationException(
 909                         local.message("ERR_Unsupported_Object_Id_Type", "long key", message));
 910             }
 911         }
 912
 913         protected String getString(Object objectId) {
 914             // get a String value from an objectId
 915             if (objectId instanceof StringId) {
 916                 return ((StringId)objectId).getId();
 917             } else if (objectId instanceof OpenJPAId) {
 918                 OpenJPAId openJPAId = (OpenJPAId)objectId;
 919                 Object id = openJPAId.getIdObject();
 920                 if (id instanceof String) {
 921                     return (String)id;
 922                 }
 923                 throw new UnsupportedOperationException(
 924                         local.message("ERR_Unsupported_Object_Id_Type", "String key", "OpenJPAId"));
 925             } else {
 926                 String message = (objectId == null)?"<null>":objectId.getClass().getName();
 927                 throw new UnsupportedOperationException(
 928                         local.message("ERR_Unsupported_Object_Id_Type", "String key", message));
 929             }
 930         }
 931
 932         protected OpenJPAStateManager getRelatedStateManager(ValueHandler handler, AbstractDomainFieldHandlerImpl fmd) {
 933             // get related object
 934             OpenJPAStateManager sm = ((NdbOpenJPAValueHandler) handler).getStateManager();
 935             NdbOpenJPAStoreManager store = ((NdbOpenJPAValueHandler) handler).getStoreManager();
 936             OpenJPAStateManager rel = RelationStrategies.getStateManager(sm.fetchObjectField(fmd.getFieldNumber()), store.getContext());
 937             return rel;
 938         }
 939
 940         public void partitionKeySetPart(AbstractDomainFieldHandlerImpl fmd,
 941                 PartitionKey partitionKey, ValueHandler keyValueHandler) {
 942             throw new ClusterJFatalInternalException(
 943                     local.message("ERR_Operation_Not_Supported","partitionKeySetPart", "non-key fields"));
 944         }
 945
 946         public Object getValue(QueryExecutionContext context, String index) {
 947             return context.getObject(index);
 948         }
 949     };
 950
 951     static ObjectOperationHandler objectOperationHandlerRelationIntField =
 952             new ObjectOperationHandlerRelationField() {
 953
 954         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, ValueHandler handler, Operation op) {
 955             OpenJPAStateManager rel = getRelatedStateManager(handler, fmd);
 956             // get openJPAId from related object
 957             if (rel == null) {
 958                 if (logger.isDetailEnabled()) logger.detail("Related object is null");
 959                 op.setNull(fmd.getStoreColumn());
 960             } else {
 961                 Object objid = rel.getObjectId();
 962                 if (objid == null) {
 963                     // TODO: doesn't seem right
 964                     op.setNull(fmd.getStoreColumn());
 965                     if (logger.isDetailEnabled()) logger.detail("Related object class: " + rel.getMetaData().getTypeAlias() + " object id: " + objid);
 966                 } else {
 967                     int oid = getInt(objid);
 968                     if (logger.isDetailEnabled()) logger.detail("Related object class: " + rel.getMetaData().getTypeAlias() + " key: " + oid);
 969                     op.setInt(fmd.getStoreColumn(), oid);
 970                 }
 971             }
 972         }
 973
 974         public String handler() {
 975             return "Object ToOne Int key.";
 976         }
 977
 978         @Override
 979         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
 980             if (value == null) {
 981                 op.setNull(fmd.getStoreColumn());
 982             } else {
 983                 op.setInt(fmd.getStoreColumn(),(Integer) value);
 984             }
 985         }
 986
 987         @Override
 988         public void filterCompareValue(AbstractDomainFieldHandlerImpl fmd, Object oid, BinaryCondition condition, ScanFilter filter) {
 989             Field field = ((NdbOpenJPADomainFieldHandlerImpl)fmd).getOidField();
 990             Object value = getKeyValue(field, oid);
 991             if (logger.isDetailEnabled()) logger.detail("For column: " + fmd.getColumnName() + " oid: " + oid + " value: " + value);
 992             filter.cmpInt(condition, fmd.getStoreColumn(), ((Integer) value).intValue());
 993         }
 994
 995         /** Set bounds for an index operation.
 996          *
 997          * @param fmd the domain field handler
 998          * @param value the value to set
 999          * @param type the bound type (i.e. EQ, NE, LE, LT, GE, GT)
1000          * @param op the index operation
1001          */
1002         public void operationSetBounds(AbstractDomainFieldHandlerImpl fmd, Object value, BoundType type, IndexScanOperation op) {
1003             op.setBoundInt(fmd.getStoreColumn(), type, (Integer)value);
1004         }
1005
1006     };
1007
1008     static ObjectOperationHandler objectOperationHandlerRelationLongField =
1009             new ObjectOperationHandlerRelationField() {
1010
1011         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, ValueHandler handler, Operation op) {
1012             OpenJPAStateManager rel = getRelatedStateManager(handler, fmd);
1013             // get openJPAId from related object
1014             if (rel == null) {
1015                 if (logger.isDetailEnabled()) logger.detail("Related object is null");
1016                 op.setNull(fmd.getStoreColumn());
1017             } else {
1018                 long oid = getLong(rel.getObjectId());
1019                 if (logger.isDetailEnabled()) logger.detail("Related object class: " + rel.getMetaData().getTypeAlias() + " key: " + oid);
1020                 op.setLong(fmd.getStoreColumn(), oid);
1021             }
1022         }
1023
1024         public String handler() {
1025             return "Object ToOne Long key.";
1026         }
1027
1028         @Override
1029         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
1030             if (value == null) {
1031                 op.setNull(fmd.getStoreColumn());
1032             } else {
1033                 op.setLong(fmd.getStoreColumn(),(Long) value);
1034             }
1035         }
1036
1037         @Override
1038         public void filterCompareValue(AbstractDomainFieldHandlerImpl fmd, Object oid, BinaryCondition condition, ScanFilter filter) {
1039             Field field = ((NdbOpenJPADomainFieldHandlerImpl)fmd).getOidField();
1040             Object value = getKeyValue(field, oid);
1041             if (logger.isDetailEnabled()) logger.detail("For column: " + fmd.getColumnName() + " oid: " + oid + " value: " + value);
1042             filter.cmpLong(condition, fmd.getStoreColumn(), ((Long) value).longValue());
1043         }
1044
1045         /** Set bounds for an index operation.
1046          *
1047          * @param fmd the domain field handler
1048          * @param value the value to set
1049          * @param type the bound type (i.e. EQ, NE, LE, LT, GE, GT)
1050          * @param op the index operation
1051          */
1052         public void operationSetBounds(AbstractDomainFieldHandlerImpl fmd, Object value, BoundType type, IndexScanOperation op) {
1053             op.setBoundLong(fmd.getStoreColumn(), type, (Long)value);
1054         }
1055
1056     };
1057
1058     static ObjectOperationHandler objectOperationHandlerRelationStringField =
1059             new ObjectOperationHandlerRelationField() {
1060
1061         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, ValueHandler handler, Operation op) {
1062             OpenJPAStateManager rel = getRelatedStateManager(handler, fmd);
1063             // get openJPAId from related object
1064             if (rel == null) {
1065                 if (logger.isDetailEnabled()) logger.detail("Related object is null");
1066                 op.setNull(fmd.getStoreColumn());
1067             } else {
1068                 String oid = getString(rel.getObjectId());
1069                 if (logger.isDetailEnabled()) logger.detail("Related object class: " + rel.getMetaData().getTypeAlias() + " key: " + oid);
1070                 op.setString(fmd.getStoreColumn(), oid);
1071             }
1072         }
1073
1074         public String handler() {
1075             return "Object ToOne String key.";
1076         }
1077
1078         @Override
1079         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
1080             if (value == null) {
1081                 op.setNull(fmd.getStoreColumn());
1082             } else {
1083                 op.setString(fmd.getStoreColumn(),(String) value);
1084             }
1085         }
1086
1087         @Override
1088         public void filterCompareValue(AbstractDomainFieldHandlerImpl fmd, Object oid, BinaryCondition condition, ScanFilter filter) {
1089             Field field = ((NdbOpenJPADomainFieldHandlerImpl)fmd).getOidField();
1090             Object value = getKeyValue(field, oid);
1091             if (logger.isDetailEnabled()) logger.detail("For column: " + fmd.getColumnName() + " oid: " + oid + " filter.cmpString: " + value);
1092             filter.cmpString(condition, fmd.getStoreColumn(), (String) value);
1093         }
1094
1095         /** Set bounds for an index operation.
1096          *
1097          * @param fmd the domain field handler
1098          * @param value the value to set
1099          * @param type the bound type (i.e. EQ, NE, LE, LT, GE, GT)
1100          * @param op the index operation
1101          */
1102         public void operationSetBounds(AbstractDomainFieldHandlerImpl fmd, Object value, BoundType type, IndexScanOperation op) {
1103             op.setBoundString(fmd.getStoreColumn(), type, (String)value);
1104         }
1105
1106     };
1107
1108     static ObjectOperationHandlerRelationField objectOperationHandlerRelationCompositeField = new ObjectOperationHandlerRelationField() {
1109
1110         public String handler() {
1111             return "Composite key.";
1112         }
1113
1114         @Override
1115         public void operationGetValue(AbstractDomainFieldHandlerImpl fmd, Operation op) {
1116             for (AbstractDomainFieldHandlerImpl localHandler: fmd.compositeDomainFieldHandlers) {
1117                 localHandler.operationGetValue(op);
1118             }
1119         }
1120
1121         @Override
1122         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, Object value, Operation op) {
1123             throw new ClusterJFatalInternalException(
1124                     local.message("ERR_Implementation_Should_Not_Occur"));
1125         }
1126
1127         /** Set the filter for this composite key relationship field. This only works for equal conditions.
1128          * Set the filter to AND, and for each field in the composite key, extract the value
1129          * from the oid and add it to the filter.
1130          */
1131         @Override
1132         public void filterCompareValue(AbstractDomainFieldHandlerImpl fmd, Object value, BinaryCondition condition, ScanFilter filter) {
1133
1134             if (!BinaryCondition.COND_EQ.equals(condition)) {
1135                 throw new ClusterJFatalInternalException(
1136                         local.message("ERR_Illegal_Filter_Condition", condition));
1137             }
1138             filter.begin();
1139             for (AbstractDomainFieldHandlerImpl localHandler : fmd.compositeDomainFieldHandlers) {
1140                 if (value != null) {
1141                     // extract the value from the oid and add the filter condition
1142                     localHandler.filterCompareValue(value, condition, filter);
1143                 } else {
1144                     // set null for each local column
1145                     localHandler.filterCompareValue((Object)null, condition, filter);
1146                 }
1147             }
1148             filter.end();
1149         }
1150
1151         public void operationSetValue(AbstractDomainFieldHandlerImpl fmd, ValueHandler handler, Operation op) {
1152             OpenJPAStateManager rel = getRelatedStateManager(handler, fmd);
1153             OpenJPAId openJPAId = null;
1154             Object oid = null;
1155             if (rel == null) {
1156                 if (logger.isDetailEnabled()) logger.detail("Related object is null");
1157             } else {
1158                 if (logger.isDetailEnabled()) logger.detail("Related object class: " + rel.getMetaData().getTypeAlias() + " key: " + openJPAId);
1159                 openJPAId = (OpenJPAId) rel.getObjectId();
1160                 oid = openJPAId.getIdObject();
1161             }
1162             for (AbstractDomainFieldHandlerImpl localHandler : fmd.compositeDomainFieldHandlers) {
1163                 Object value = null;
1164                 if (rel != null) {
1165                     // get the value from the related object
1166                     Field field = ((NdbOpenJPADomainFieldHandlerImpl)localHandler).getOidField();
1167                     value = getKeyValue(field, oid);
1168                     localHandler.operationSetValue(value, op);
1169                 } else {
1170                     // set null for each local column
1171                     localHandler.operationSetValue((Object)null, op);
1172                 }
1173             }
1174         }
1175
1176         @Override
1177         public Map<String, Object> createParameterMap(NdbOpenJPADomainFieldHandlerImpl domainFieldHandler,
1178                 QueryDomainType<?> queryDomainObject, Object oid) {
1179             Map<String, Object> result = new HashMap<String, Object>();
1180             Predicate predicate = null;
1181             for (AbstractDomainFieldHandlerImpl localHandler: domainFieldHandler.compositeDomainFieldHandlers) {
1182                 String name = localHandler.getColumnName();
1183                 PredicateOperand parameter = queryDomainObject.param(name);
1184                 PredicateOperand field = queryDomainObject.get(name);
1185                 if (predicate == null) {
1186                     predicate = field.equal(parameter);
1187                 } else {
1188                     predicate.and(field.equal(parameter));
1189                 }
1190                 // construct a map of parameter binding to the value in oid
1191                 Object value = domainFieldHandler.getKeyValue(oid);
1192                 result.put(name, value);
1193                 if (logger.isDetailEnabled()) logger.detail("Map.Entry key: " + name + ", value: " + value);
1194             }
1195             queryDomainObject.where(predicate);
1196             return result;
1197         }
1198
1199         /** Set bounds for an index operation. Delegate this to each domain field handler which
1200          * will extract the column data from the object id.
1201          *
1202          * @param fmd the domain field handler
1203          * @param oid the value to set
1204          * @param type the bound type (i.e. EQ, LE, LT, GE, GT)
1205          * @param op the index operation
1206          */
1207         public void operationSetBounds(AbstractDomainFieldHandlerImpl fmd, Object oid, BoundType type, IndexScanOperation op) {
1208             for (AbstractDomainFieldHandlerImpl localHandler : fmd.compositeDomainFieldHandlers) {
1209                 Field field = ((NdbOpenJPADomainFieldHandlerImpl)localHandler).getOidField();
1210                 Object columnData = getKeyValue(field, oid);
1211                 localHandler.operationSetBounds(columnData, type, op);
1212             }
1213         }
1214
1215     };
1216
1217     public com.mysql.clusterj.core.store.Column[] getStoreColumns() {
1218         return storeColumns;
1219     }
1220
1221     public boolean isSupported() {
1222         return supported;
1223     }
1224
1225     public boolean isRelation() {
1226         return isRelation;
1227     }
1228
1229     public String getReason() {
1230         return reason;
1231     }
1232
1233     private void setUnsupported(String reason) {
1234         this.supported = false;
1235         this.reason = reason;
1236     }
1237
1238 }