added jdbc source type parameter

Author: Jolanrensen

dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DbType.kt

@@ -196,6 +196,7 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
             }
 
         return typeInformationWithPostprocessingFor(
+            jdbcSourceType = kType.withNullability(tableColumnMetadata.isNullable),
             targetSchema = ColumnSchema.Value(kType.withNullability(tableColumnMetadata.isNullable)),
             columnPostprocessor = postprocessor?.castToAny(),
         )
@@ -242,6 +243,8 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
                     type = schema.type,
                 )
 
+            // TODO, this should be postponed to post-processing.
+            //  List<AnyRow>.toDataFrame() is heavy!
             is ColumnSchema.Group ->
                 DataColumn.createColumnGroup(
                     name = name,

dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DuckDb.kt

@@ -59,6 +59,7 @@ import kotlin.collections.toList
 import kotlin.reflect.KTypeProjection
 import kotlin.reflect.full.createType
 import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
 import kotlin.time.Instant
 import kotlin.time.toKotlinInstant
 import kotlin.uuid.Uuid
@@ -170,6 +171,7 @@ public object DuckDb : DbType("duckdb") {
                     ).withNullability(isNullable)
 
                     typeInformationWithPreprocessingForValueColumnOf<Map<String, Any?>, Map<String, Any?>>(
+                        jdbcSourceType = typeOf<Map<String, Any?>>().withNullability(isNullable), // unused
                         targetColumnType = targetMapType,
                     ) { map, _ ->
                         // only need to preprocess the values, as the keys are just Strings
@@ -195,6 +197,7 @@ public object DuckDb : DbType("duckdb") {
 
                     // todo maybe List<DataRow> should become FrameColumn
                     typeInformationWithPreprocessingFor<SqlArray, List<Any?>>(
+                        jdbcSourceType = typeOf<SqlArray>().withNullability(isNullable),
                         targetSchema = ColumnSchema.Value(targetListType),
                     ) { sqlArray, _ ->
                         sqlArray
@@ -204,7 +207,11 @@ public object DuckDb : DbType("duckdb") {
                 }
 
                 // TODO requires #1266 for specific types
-                STRUCT -> typeInformationForValueColumnOf<Struct>(isNullable)
+                STRUCT -> {
+                    val structTypes = parseStructType(sqlTypeName)
+
+                    typeInformationForValueColumnOf<Struct>(isNullable)
+                }
 
                 // Cannot handle this in Kotlin
                 UNION -> typeInformationForValueColumnOf<Any>(isNullable)
@@ -269,6 +276,17 @@ public object DuckDb : DbType("duckdb") {
         return typeString.take(typeString.indexOfLast { it == '[' })
     }
 
+    /** Parses "STRUCT(v VARCHAR, i INTEGER)" into [("v", "VARCHAR"), ("i", "INTEGER")] */
+    internal fun parseStructType(typeString: String): Map<String, String> {
+        if (!typeString.startsWith("STRUCT(")) {
+            error("invalid STRUCT type: $typeString")
+        }
+        return typeString.removeSurrounding("STRUCT(", ")")
+            .split(",")
+            .map { it.trim().split(" ") }
+            .associate { (name, type) -> name to type }
+    }
+
     /**
      * How to filter out system tables from user-created ones when using
      * [DataFrame.readAllSqlTables][DataFrame.Companion.readAllSqlTables] and

dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/TypeInformation.kt

@@ -28,6 +28,7 @@ public typealias AnyTypeInformation = TypeInformation<*, *, *>
  *   to a [DataColumn] of with values of type [P].
  */
 public open class TypeInformation<J : Any, D : Any, P : Any>(
+    public open val jdbcSourceType: KType,
     public open val targetSchema: ColumnSchema,
     public open val valuePreprocessor: DbValuePreprocessor<J, D>?,
     public open val columnPostprocessor: DbColumnPostprocessor<D, P>?,
@@ -55,38 +56,45 @@ public fun TypeInformation<*, *, *>.castToAny(): TypeInformation<Any, Any, Any>
 // region generic constructors
 
 public fun <J : Any, D : Any, P : Any> typeInformationWithProcessingFor(
+    jdbcSourceType: KType,
     targetSchema: ColumnSchema,
     valuePreprocessor: DbValuePreprocessor<J, D>?,
     columnPostprocessor: DbColumnPostprocessor<D, P>?,
 ): TypeInformation<J, D, P> =
     TypeInformation(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = targetSchema,
         valuePreprocessor = valuePreprocessor,
         columnPostprocessor = columnPostprocessor,
     )
 
-public fun <J : Any> typeInformationFor(targetSchema: ColumnSchema): TypeInformation<J, J, J> =
+public fun <J : Any> typeInformationFor(jdbcSourceType: KType, targetSchema: ColumnSchema): TypeInformation<J, J, J> =
     typeInformationWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = targetSchema,
         valuePreprocessor = null,
         columnPostprocessor = null,
     )
 
 public fun <J : Any, D : Any> typeInformationWithPreprocessingFor(
+    jdbcSourceType: KType,
     targetSchema: ColumnSchema,
     valuePreprocessor: DbValuePreprocessor<J, D>?,
 ): TypeInformation<J, D, D> =
     typeInformationWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = targetSchema,
         valuePreprocessor = valuePreprocessor,
         columnPostprocessor = null,
     )
 
 public fun <J : Any, P : Any> typeInformationWithPostprocessingFor(
+    jdbcSourceType: KType,
     targetSchema: ColumnSchema,
     columnPostprocessor: DbColumnPostprocessor<J, P>?,
 ): TypeInformation<J, J, P> =
     typeInformationWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = targetSchema,
         valuePreprocessor = null,
         columnPostprocessor = columnPostprocessor,
@@ -97,43 +105,49 @@ public fun <J : Any, P : Any> typeInformationWithPostprocessingFor(
 // region ValueColumn constructors
 
 public fun <J : Any> typeInformationForValueColumnOf(kType: KType): TypeInformation<J, J, J> =
-    typeInformationFor(targetSchema = ColumnSchema.Value(kType))
+    typeInformationFor(jdbcSourceType = kType, targetSchema = ColumnSchema.Value(kType))
 
 public inline fun <reified J : Any> typeInformationForValueColumnOf(isNullable: Boolean): TypeInformation<J, J, J> =
     typeInformationForValueColumnOf(typeOf<J>().withNullability(isNullable))
 
 public fun <J : Any, D : Any> typeInformationWithPreprocessingForValueColumnOf(
+    jdbcSourceType: KType,
     targetColumnType: KType,
     valuePreprocessor: DbValuePreprocessor<J, D>?,
 ): TypeInformation<J, D, D> =
     typeInformationWithPreprocessingFor(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = ColumnSchema.Value(targetColumnType),
         valuePreprocessor = valuePreprocessor,
     )
 
-public inline fun <J : Any, reified D : Any> typeInformationWithPreprocessingForValueColumnOf(
+public inline fun <reified J : Any, reified D : Any> typeInformationWithPreprocessingForValueColumnOf(
     isNullable: Boolean,
     valuePreprocessor: DbValuePreprocessor<J, D>?,
 ): TypeInformation<J, D, D> =
     typeInformationWithPreprocessingForValueColumnOf(
+        jdbcSourceType = typeOf<J>().withNullability(isNullable),
         targetColumnType = typeOf<D>().withNullability(isNullable),
         valuePreprocessor = valuePreprocessor,
     )
 
 public fun <J : Any, P : Any> typeInformationWithPostprocessingForValueColumnOf(
+    jdbcSourceType: KType,
     targetColumnType: KType,
     columnPostprocessor: DbColumnPostprocessor<J, P>?,
 ): TypeInformation<J, J, P> =
     typeInformationWithPostprocessingFor(
+        jdbcSourceType = jdbcSourceType,
         targetSchema = ColumnSchema.Value(targetColumnType),
         columnPostprocessor = columnPostprocessor,
     )
 
-public inline fun <J : Any, reified P : Any> typeInformationWithPostprocessingForValueColumnOf(
+public inline fun <reified J : Any, reified P : Any> typeInformationWithPostprocessingForValueColumnOf(
     isNullable: Boolean,
     columnPostprocessor: DbColumnPostprocessor<J, P>?,
 ): TypeInformation<J, J, P> =
     typeInformationWithPostprocessingForValueColumnOf(
+        jdbcSourceType = typeOf<J>().withNullability(isNullable),
         targetColumnType = typeOf<P>().withNullability(isNullable),
         columnPostprocessor = columnPostprocessor,
     )