feat: follow key chains when resolving foreign key columns

docs/relationships.md

@@ -463,6 +463,34 @@ Storm extracts the primary key from the `User` entity and uses it as the value f
 </TabItem>
 </Tabs>
 
+### Key Chains
+
+The referenced entity's primary key may itself be a foreign key — or a compound key record. Storm follows this *key chain* to its terminal columns. A dependent one-to-one on an entity that is itself a dependent one-to-one works the same way as the single-level case:
+
+<Tabs groupId="language">
+<TabItem value="kotlin" label="Kotlin" default>
+
+```kotlin
+data class ProfileAudit(
+    @PK(generation = NONE) @FK val profile: UserProfile,  // UserProfile's own PK is the FK to User
+    val remark: String
+) : Entity<UserProfile>
+```
+
+</TabItem>
+<TabItem value="java" label="Java">
+
+```java
+record ProfileAudit(@PK(generation = NONE) @FK UserProfile profile,  // UserProfile's own PK is the FK to User
+                    String remark
+) implements Entity<UserProfile> {}
+```
+
+</TabItem>
+</Tabs>
+
+The foreign key spans the same columns as the terminal key of the chain: a single-column chain resolves to one column named by the FK convention (`profile_id` here), and a compound key contributes the referenced key's column names. Circular key chains are rejected at model construction with a clear error.
+
 ---
 
 ## Relationship Loading Behavior

storm-core/src/main/java/st/orm/core/repository/impl/MergeEntityRepositoryImpl.java

@@ -62,7 +62,7 @@
  * FROM clause (Oracle), {@link #statementSuffix()} for dialects that require a statement terminator (SQL Server),
  * the version expressions, and {@link #mergeInsert()} for the dialect-specific WHEN NOT MATCHED clause.</p>
  *
- * @since 1.12
+ * @since 1.11
  */
 public abstract class MergeEntityRepositoryImpl<E extends Entity<ID>, ID> extends EntityRepositoryImpl<E, ID> {
 

storm-core/src/main/java/st/orm/core/spi/JsonString.java

@@ -26,7 +26,7 @@
  * {@code jsonb} columns.</p>
  *
  * @param value the serialized JSON text.
- * @since 1.12
+ * @since 1.11
  */
 public record JsonString(@Nonnull String value) {
 

storm-core/src/main/java/st/orm/core/template/Column.java

@@ -49,10 +49,26 @@ public interface Column {
     /**
      * Gets the type of the column.
      *
+     * <p>This is the declared field type: for foreign key columns the referenced entity type, for
+     * converter-backed columns the converted field type's parameter type, and the field type otherwise.</p>
+     *
      * @return the type of the column.
      */
     Class<?> type();
 
+    /**
+     * Gets the Java type of the value as it is persisted to this column.
+     *
+     * <p>Unlike {@link #type()}, this is the terminal type after following key chains: for a foreign key
+     * column it is the type of the referenced key's terminal field (following foreign keys that are
+     * themselves primary keys, and record key components), letting callers reason about the physical column
+     * without knowledge of key structure. For all other columns it equals {@link #type()}.</p>
+     *
+     * @return the Java type persisted to this column.
+     * @since 1.11
+     */
+    Class<?> persistedType();
+
     /**
      * Determines if the column is a primary key.
      *

storm-core/src/main/java/st/orm/core/template/SqlDialect.java

@@ -403,7 +403,7 @@ default void setParameter(@Nonnull PreparedStatement preparedStatement, int inde
      * @param index the parameter index.
      * @param json the serialized JSON value.
      * @throws SQLException if a database access error occurs.
-     * @since 1.12
+     * @since 1.11
      */
     default void setParameter(@Nonnull PreparedStatement preparedStatement, int index,
                               @Nonnull JsonString json) throws SQLException {

storm-core/src/main/java/st/orm/core/template/impl/ColumnImpl.java

@@ -47,6 +47,7 @@ public record ColumnImpl(
         @Nonnull Name columnName,
         int index,
         @Nonnull Class<?> type,
+        @Nonnull Class<?> persistedType,
         boolean primaryKey,
         @Nonnull GenerationStrategy generation,
         @Nonnull String sequence,

storm-core/src/main/java/st/orm/core/template/impl/ModelFactory.java

@@ -24,6 +24,7 @@
 import static st.orm.core.template.impl.RecordReflection.getColumnName;
 import static st.orm.core.template.impl.RecordReflection.getDiscriminatorColumn;
 import static st.orm.core.template.impl.RecordReflection.getDiscriminatorColumnJavaType;
+import static st.orm.core.template.impl.RecordReflection.getFkLeaves;
 import static st.orm.core.template.impl.RecordReflection.getForeignKeys;
 import static st.orm.core.template.impl.RecordReflection.getGenerationStrategy;
 import static st.orm.core.template.impl.RecordReflection.getRecordType;
@@ -130,7 +131,7 @@ private static List<Column> adjustColumnsForJoinedSubtype(@Nonnull List<Column>
                 if (ci.primaryKey()) {
                     // PK: override generation to NONE (PK is provided from base INSERT).
                     adjusted.add(new ColumnImpl(
-                            ci.columnName(), ci.index(), ci.type(), true,
+                            ci.columnName(), ci.index(), ci.type(), ci.persistedType(), true,
                             GenerationStrategy.NONE, "",
                             ci.foreignKey(), ci.foreignKeyGeneration(), ci.keyIndex(), ci.nullable(),
                             ci.insertable(), ci.updatable(), ci.version(), ci.ref(),
@@ -147,7 +148,7 @@ private static List<Column> adjustColumnsForJoinedSubtype(@Nonnull List<Column>
                     if (isBaseField) {
                         // Base non-PK field: not insertable or updatable in extension table.
                         adjusted.add(new ColumnImpl(
-                                ci.columnName(), ci.index(), ci.type(), false,
+                                ci.columnName(), ci.index(), ci.type(), ci.persistedType(), false,
                                 ci.generation(), ci.sequence(),
                                 ci.foreignKey(), ci.foreignKeyGeneration(), ci.keyIndex(), ci.nullable(),
                                 false, false, ci.version(), ci.ref(),
@@ -204,6 +205,7 @@ private static <T extends Data, ID> Model<T, ID> createSealedModel(@Nonnull Mode
                 new ColumnName(discriminatorColumnName, false),
                 index.getAndIncrement(),
                 getDiscriminatorColumnJavaType(sealedType),
+                getDiscriminatorColumnJavaType(sealedType),
                 false,  // not primary key
                 GenerationStrategy.NONE,
                 "",
@@ -295,6 +297,7 @@ private static <T extends Data, ID> Model<T, ID> createSealedModel(@Nonnull Mode
                     columnName,
                     index.getAndIncrement(),
                     field.type(),
+                    field.type(),
                     isPk,
                     isPk ? getGenerationStrategy(field) : GenerationStrategy.NONE,
                     isPk ? getSequence(field) : "",
@@ -398,7 +401,7 @@ private static void createColumns(@Nonnull BuildContext ctx,
                     throw new SqlTemplateException("Column count does not match value count.");
                 }
                 ColumnSpec spec = buildSpec(field, effectivePrimaryKey, fkAnnotation, parentNullable, inheritedPersist, pkContext);
-                emitColumns(ctx, field, columnMetamodel, null, spec, keyScope, columnNames, columnTypes);
+                emitColumns(ctx, field, columnMetamodel, null, spec, keyScope, columnNames, columnTypes, columnTypes);
                 return;
             }
             if (isRecord(field.type()) && !fkAnnotation) {
@@ -434,14 +437,18 @@ private static void createColumns(@Nonnull BuildContext ctx,
                             (Class<? extends Data>) ownMetamodel.root(),
                             ownMetamodel.fieldPath().isEmpty() ? field.name() : ownMetamodel.fieldPath() + "." + pkField.name());
                 }
-                emitColumns(ctx, field, ownMetamodel, secondaryMetamodel, spec, keyScope, fkNames, nCopies(fkNames.size(), spec.dataType()));
+                var leaves = getFkLeaves(field);
+                List<Class<?>> persistedTypes = leaves != null && leaves.size() == fkNames.size()
+                        ? leaves.stream().<Class<?>>map(leaf -> leaf.field().type()).toList()
+                        : nCopies(fkNames.size(), spec.dataType());
+                emitColumns(ctx, field, ownMetamodel, secondaryMetamodel, spec, keyScope, fkNames, nCopies(fkNames.size(), spec.dataType()), persistedTypes);
                 if (!spec.ref()) {
                     expandForeignRelation(ctx, ownMetamodel, field, spec.nullable());
                 }
                 return;
             }
             ColumnName columnName = getColumnName(field, ctx.builder().columnNameResolver());
-            emitColumns(ctx, field, columnMetamodel, null, spec, keyScope, List.of(columnName), List.of(spec.dataType()));
+            emitColumns(ctx, field, columnMetamodel, null, spec, keyScope, List.of(columnName), List.of(spec.dataType()), List.of(spec.dataType()));
         } catch (SqlTemplateException e) {
             throw new UncheckedSqlTemplateException(e);
         }
@@ -499,8 +506,9 @@ private static void emitColumns(@Nonnull BuildContext ctx,
                                     @Nonnull ColumnSpec spec,
                                     @Nonnull KeyScope keyScope,
                                     @Nonnull List<? extends Name> names,
-                                    @Nonnull List<Class<?>> types) throws SqlTemplateException {
-        if (names.size() != types.size()) {
+                                    @Nonnull List<Class<?>> types,
+                                    @Nonnull List<Class<?>> persistedTypes) throws SqlTemplateException {
+        if (names.size() != types.size() || names.size() != persistedTypes.size()) {
             throw new SqlTemplateException("Column count does not match type count.");
         }
         for (int i = 0; i < names.size(); i++) {
@@ -509,6 +517,7 @@ private static void emitColumns(@Nonnull BuildContext ctx,
                     names.get(i),
                     ctx.index().getAndIncrement(),
                     types.get(i),
+                    persistedTypes.get(i),
                     spec.primaryKey(),
                     spec.generation(),
                     spec.sequence(),

storm-core/src/main/java/st/orm/core/template/impl/ModelImpl.java

@@ -25,6 +25,7 @@
 import static st.orm.core.spi.Providers.getORMConverter;
 import static st.orm.core.template.impl.ObjectMapperFactory.nullableHint;
 import static st.orm.core.template.impl.RecordReflection.getDiscriminatorValue;
+import static st.orm.core.template.impl.RecordReflection.getFkLeaves;
 import static st.orm.core.template.impl.RecordReflection.getRefDataType;
 import static st.orm.core.template.impl.RecordReflection.isJoinedEntity;
 import static st.orm.core.template.impl.RecordReflection.isPolymorphicData;
@@ -131,6 +132,14 @@ public final class ModelImpl<E extends Data, ID> implements Model<E, ID> {
      */
     private final List<ORMConverter> converters;
 
+    /**
+     * Per-column accessor paths for foreign key columns, following the referenced key chain from the foreign
+     * key field's value down to the terminal field bound to the column. Precomputed at model construction so
+     * value extraction is a plain loop of accessor invocations; {@code null} for columns that do not use chain
+     * extraction (non-foreign-key columns and polymorphic foreign keys).
+     */
+    private final RecordField[][] keyPaths;
+
     /**
      * Parent metamodel for each converter group, keyed by the group's 0-based start column index. A converter reads
      * its value by reflectively invoking the field accessor on the record passed to
@@ -177,6 +186,7 @@ public ModelImpl(@Nonnull RecordType recordType,
         this.polymorphicFkDiscriminatorColumns = initPolymorphicFkDiscriminatorColumns(this.fields, this.columns);
         this.converters = initConverters(this.fields, this.columns);
         this.converterParents = initConverterParents(this.converters, this.columns);
+        this.keyPaths = initKeyPaths(this.fields, this.columns);
         this.primaryKeyField = initPrimaryKeyField(this.recordType);
         this.declaredColumns = initDeclaredColumns(this.columns);
         this.primaryKeyMetamodel = initPrimaryKeyMetamodel(this.declaredColumns);
@@ -236,6 +246,63 @@ private static Map<Integer, Class<?>> initPolymorphicFkDiscriminatorColumns(
         return map.isEmpty() ? Map.of() : Map.copyOf(map);
     }
 
+    private static RecordField[][] initKeyPaths(@Nonnull List<RecordField> fields,
+                                                @Nonnull List<Column> columns) throws SqlTemplateException {
+        var paths = new RecordField[columns.size()][];
+        Map<RecordField, Integer> columnsByField = new HashMap<>();
+        for (int i = 0; i < columns.size(); i++) {
+            if (columns.get(i).foreignKey()) {
+                columnsByField.merge(fields.get(i), 1, Integer::sum);
+            }
+        }
+        Map<RecordField, Optional<List<RecordReflection.KeyLeaf>>> leavesByField = new HashMap<>();
+        for (int i = 0; i < columns.size(); i++) {
+            var column = columns.get(i);
+            if (!column.foreignKey()) {
+                continue;
+            }
+            var field = fields.get(i);
+            var cached = leavesByField.get(field);
+            if (cached == null) {
+                cached = Optional.ofNullable(getFkLeaves(field));
+                leavesByField.put(field, cached);
+            }
+            // Only use chain extraction when the flattened key matches the emitted columns one-to-one;
+            // polymorphic foreign keys keep the legacy extraction.
+            var leaves = cached.orElse(null);
+            if (leaves == null || leaves.size() != columnsByField.get(field)) {
+                continue;
+            }
+            int keyIndex = column.keyIndex();
+            if (keyIndex >= 1 && keyIndex <= leaves.size()) {
+                paths[i] = leaves.get(keyIndex - 1).path().toArray(new RecordField[0]);
+            }
+        }
+        return paths;
+    }
+
+    /**
+     * Extracts the value bound to a foreign key column by walking the column's key chain path. {@link Ref}
+     * values substitute their identifier for the next step of the path.
+     */
+    @Nullable
+    private Object extractKeyValue(@Nullable Object value,
+                                   @Nonnull RecordField[] path,
+                                   @Nonnull Column column) throws SqlTemplateException {
+        if (value == null) {
+            return null;
+        }
+        if (!(value instanceof Ref) && !(value instanceof Data)) {
+            throw new SqlTemplateException("Invalid foreign key type for column '%s'. Expected a Ref or Data type, but got an unrecognized value type. Ensure the foreign key field is correctly typed.".formatted(column.name()));
+        }
+        for (int i = 0; i < path.length && value != null; i++) {
+            value = value instanceof Ref<?> ref
+                    ? ref.id()                          // The identifier is the result of the next accessor.
+                    : REFLECTION.invoke(path[i], value);
+        }
+        return value;
+    }
+
     private static List<ORMConverter> initConverters(List<RecordField> fields, List<Column> columns) {
         var converters = new ORMConverter[columns.size()];
         for (int i = 0; i < columns.size(); i++) {
@@ -487,6 +554,10 @@ public void forEachValue(@Nonnull Metamodel<E, ?> metamodel,
         Object value;
         if (object instanceof Data data) {
             value = REFLECTION.getId(data);
+            // Follow the key chain: the identifier of an entity-keyed entity is itself an entity.
+            while (value instanceof Data nested) {
+                value = REFLECTION.getId(nested);
+            }
         } else {
             value = object;
         }
@@ -549,12 +620,21 @@ private void forEachInlineValue(@Nonnull Metamodel<E, ?> metamodel,
                         .formatted(componentName, object.getClass().getSimpleName()));
             }
             Object value = REFLECTION.getRecordValue(object, componentIndex);
-            // Handle FK components: extract the primary key ID from the Data object.
-            if (column.foreignKey() && value instanceof Data data) {
-                value = REFLECTION.getId(data);
-            }
-            // Handle compound keys (e.g., compound FK or compound PK within the inline record).
-            if (value != null && (column.primaryKey() || column.foreignKey()) && isRecord(value.getClass())) {
+            if (column.foreignKey()) {
+                // Handle FK components: walk the key chain from the referenced Data object to the column value.
+                var path = keyPaths[column.index() - 1];
+                if (path != null) {
+                    value = extractKeyValue(value, path, column);
+                } else {
+                    if (value instanceof Data data) {
+                        value = REFLECTION.getId(data);
+                    }
+                    if (value != null && isRecord(value.getClass())) {
+                        value = REFLECTION.getRecordValue(value, column.keyIndex() - 1);
+                    }
+                }
+            } else if (value != null && column.primaryKey() && isRecord(value.getClass())) {
+                // Handle compound primary keys within the inline record.
                 value = REFLECTION.getRecordValue(value, column.keyIndex() - 1);
             }
             consumer.accept(column, map(column, value));
@@ -627,20 +707,28 @@ private void forEachValueOrdered(@Nonnull List<Column> view,
                 throw new SqlTemplateException("Cannot write NULL to non-nullable column '%s'. Ensure the entity field has a value before inserting or updating, or %s if NULL is intended.".formatted(column.name(), nullableHint(type())));
             }
             if (column.foreignKey()) {
-                if (value instanceof Ref<?> ref) {
-                    Class<?> polymorphicSealedType = polymorphicFkDiscriminatorColumns.get(column.index());
+                if (value instanceof Ref<?> ref && polymorphicFkDiscriminatorColumns.containsKey(column.index())) {
                     // Polymorphic FK discriminator column: extract the discriminator value from the Ref's
                     // concrete type instead of the FK id.
-                    value = polymorphicSealedType != null
-                            ? getDiscriminatorValue(ref.type(), polymorphicSealedType)
-                            : ref.id();
-                } else if (value instanceof Data data) {
-                    value = REFLECTION.getId(data);
-                } else if (value != null) {
-                    throw new SqlTemplateException("Invalid foreign key type for column '%s'. Expected a Ref or Data type, but got an unrecognized value type. Ensure the foreign key field is correctly typed.".formatted(column.name()));
+                    value = getDiscriminatorValue(ref.type(), polymorphicFkDiscriminatorColumns.get(column.index()));
+                } else {
+                    var path = keyPaths[index - 1];
+                    if (path != null) {
+                        value = extractKeyValue(value, path, column);
+                    } else {
+                        if (value instanceof Ref<?> ref) {
+                            value = ref.id();
+                        } else if (value instanceof Data data) {
+                            value = REFLECTION.getId(data);
+                        } else if (value != null) {
+                            throw new SqlTemplateException("Invalid foreign key type for column '%s'. Expected a Ref or Data type, but got an unrecognized value type. Ensure the foreign key field is correctly typed.".formatted(column.name()));
+                        }
+                        if (value != null && isRecord(value.getClass())) {
+                            value = REFLECTION.getRecordValue(value, column.keyIndex() - 1);
+                        }
+                    }
                 }
-            }
-            if (value != null && (column.primaryKey() || column.foreignKey()) && isRecord(value.getClass())) {
+            } else if (value != null && column.primaryKey() && isRecord(value.getClass())) {
                 value = REFLECTION.getRecordValue(value, column.keyIndex() - 1);
             }
             consumer.accept(column, map(column, value));