diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/AdvancedDbType.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/AdvancedDbType.kt
new file mode 100644
index 0000000000..31c824b725
--- /dev/null
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/AdvancedDbType.kt
@@ -0,0 +1,76 @@
+package org.jetbrains.kotlinx.dataframe.io.db
+
+import org.jetbrains.kotlinx.dataframe.DataColumn
+import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
+import java.sql.ResultSet
+import kotlin.reflect.KType
+
+/**
+ * Alternative version of [DbType] that allows to customize type mapping
+ * by initializing a [JdbcToDataFrameConverter] instance for each JDBC type.
+ *
+ * This can be helpful for JDBC databases that support structured data, like [DuckDb]
+ * or that need to a lot of type mapping.
+ */
+public abstract class AdvancedDbType(dbTypeInJdbcUrl: String) : DbType(dbTypeInJdbcUrl) {
+
+    protected abstract fun generateConverter(tableColumnMetadata: TableColumnMetadata): AnyJdbcToDataFrameConverter
+
+    private val converterCache = mutableMapOf<TableColumnMetadata, AnyJdbcToDataFrameConverter>()
+
+    protected fun getConverter(tableColumnMetadata: TableColumnMetadata): AnyJdbcToDataFrameConverter =
+        converterCache.getOrPut(tableColumnMetadata) {
+            generateConverter(tableColumnMetadata)
+        }
+
+    final override fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType =
+        getConverter(tableColumnMetadata).expectedJdbcType
+
+    final override fun getPreprocessedValueType(
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+    ): KType = getConverter(tableColumnMetadata).preprocessedValueType
+
+    final override fun getTargetColumnSchema(
+        tableColumnMetadata: TableColumnMetadata,
+        expectedValueType: KType,
+    ): ColumnSchema = getConverter(tableColumnMetadata).targetSchema
+
+    final override fun <J : Any> getValueFromResultSet(
+        rs: ResultSet,
+        columnIndex: Int,
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+    ): J? =
+        getConverter(tableColumnMetadata).cast<J, Any, Any>()
+            .getValueFromResultSetOrElse(rs, columnIndex) {
+                try {
+                    rs.getObject(columnIndex + 1)
+                } catch (_: Throwable) {
+                    // TODO?
+                    rs.getString(columnIndex + 1)
+                } as J?
+            }
+
+    final override fun <J : Any, D : Any> preprocessValue(
+        value: J?,
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+        expectedPreprocessedValueType: KType,
+    ): D? = getConverter(tableColumnMetadata).cast<J, D, Any>().preprocessOrCast(value)
+
+    final override fun <D : Any, P : Any> buildDataColumn(
+        name: String,
+        values: List<D?>,
+        tableColumnMetadata: TableColumnMetadata,
+        targetColumnSchema: ColumnSchema,
+        inferNullability: Boolean,
+    ): DataColumn<P?> =
+        getConverter(tableColumnMetadata).cast<Any, D, P>()
+            .buildDataColumnOrNull(name, values, inferNullability)
+            ?: values.toDataColumn(
+                name = name,
+                targetColumnSchema = targetColumnSchema,
+                inferNullability = inferNullability,
+            )
+}
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DbType.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DbType.kt
index e4fb482d8f..3da18fe5dd 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DbType.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DbType.kt
@@ -1,7 +1,15 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
+import kotlinx.datetime.LocalDateTime
+import kotlinx.datetime.toKotlinLocalDateTime
+import org.jetbrains.kotlinx.dataframe.AnyFrame
+import org.jetbrains.kotlinx.dataframe.AnyRow
 import org.jetbrains.kotlinx.dataframe.DataColumn
 import org.jetbrains.kotlinx.dataframe.api.Infer
+import org.jetbrains.kotlinx.dataframe.api.asDataColumn
+import org.jetbrains.kotlinx.dataframe.api.cast
+import org.jetbrains.kotlinx.dataframe.api.schema
+import org.jetbrains.kotlinx.dataframe.api.toDataFrame
 import org.jetbrains.kotlinx.dataframe.io.DbConnectionConfig
 import org.jetbrains.kotlinx.dataframe.io.readAllSqlTables
 import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
@@ -21,17 +29,21 @@ import java.sql.SQLXML
 import java.sql.Time
 import java.sql.Timestamp
 import java.sql.Types
-import java.time.LocalDateTime
 import java.time.OffsetDateTime
 import java.time.OffsetTime
-import java.util.Date
 import java.util.UUID
+import kotlin.collections.toTypedArray
 import kotlin.reflect.KClass
 import kotlin.reflect.KType
-import kotlin.reflect.full.createType
-import kotlin.reflect.full.isSupertypeOf
 import kotlin.reflect.full.safeCast
-import kotlin.reflect.full.starProjectedType
+import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
+import kotlin.time.Instant
+import kotlin.time.toKotlinInstant
+import kotlin.uuid.Uuid
+import kotlin.uuid.toKotlinUuid
+import java.time.LocalDateTime as JavaLocalDateTime
+import java.util.Date as JavaDate
 
 /**
  * The `DbType` class represents a database type used for reading dataframe from the database.
@@ -39,6 +51,7 @@ import kotlin.reflect.full.starProjectedType
  * @property [dbTypeInJdbcUrl] The name of the database as specified in the JDBC URL.
  */
 public abstract class DbType(public val dbTypeInJdbcUrl: String) {
+
     /**
      * Represents the JDBC driver class name for a given database type.
      *
@@ -82,10 +95,210 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
      */
     public open val defaultQueryTimeout: Int? = null // null = no timeout
 
+    /** Default mapping of [Java SQL Types][Types] to [KType]. */
+    protected val defaultJdbcTypeToKTypeMapping: Map<Int, KType> = mapOf(
+        Types.BIT to typeOf<Boolean>(),
+        Types.TINYINT to typeOf<Int>(),
+        Types.SMALLINT to typeOf<Int>(),
+        Types.INTEGER to typeOf<Int>(),
+        Types.BIGINT to typeOf<Long>(),
+        Types.FLOAT to typeOf<Float>(),
+        Types.REAL to typeOf<Float>(),
+        Types.DOUBLE to typeOf<Double>(),
+        Types.NUMERIC to typeOf<BigDecimal>(),
+        Types.DECIMAL to typeOf<BigDecimal>(),
+        Types.CHAR to typeOf<String>(),
+        Types.VARCHAR to typeOf<String>(),
+        Types.LONGVARCHAR to typeOf<String>(),
+        Types.DATE to typeOf<JavaDate>(),
+        Types.TIME to typeOf<Time>(),
+        Types.TIMESTAMP to typeOf<Timestamp>(),
+        Types.BINARY to typeOf<ByteArray>(),
+        Types.VARBINARY to typeOf<ByteArray>(),
+        Types.LONGVARBINARY to typeOf<ByteArray>(),
+        Types.NULL to typeOf<String>(),
+        Types.JAVA_OBJECT to typeOf<Any>(),
+        Types.DISTINCT to typeOf<Any>(),
+        Types.STRUCT to typeOf<Any>(),
+        Types.ARRAY to typeOf<Array<*>>(),
+        Types.BLOB to typeOf<ByteArray>(),
+        Types.CLOB to typeOf<Clob>(),
+        Types.REF to typeOf<Ref>(),
+        Types.DATALINK to typeOf<Any>(),
+        Types.BOOLEAN to typeOf<Boolean>(),
+        Types.ROWID to typeOf<RowId>(),
+        Types.NCHAR to typeOf<String>(),
+        Types.NVARCHAR to typeOf<String>(),
+        Types.LONGNVARCHAR to typeOf<String>(),
+        Types.NCLOB to typeOf<NClob>(),
+        Types.SQLXML to typeOf<SQLXML>(),
+        Types.REF_CURSOR to typeOf<Ref>(),
+        Types.TIME_WITH_TIMEZONE to typeOf<OffsetTime>(),
+        Types.TIMESTAMP_WITH_TIMEZONE to typeOf<OffsetDateTime>(),
+    )
+
+    public open fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType {
+        val kType = when (tableColumnMetadata.jdbcType) {
+            Types.OTHER ->
+                when (tableColumnMetadata.javaClassName) {
+                    "[B" -> typeOf<ByteArray>()
+                    else -> typeOf<Any>()
+                }
+
+            Types.TIMESTAMP if tableColumnMetadata.javaClassName == "java.time.LocalDateTime" ->
+                typeOf<JavaLocalDateTime>()
+
+            Types.BINARY if tableColumnMetadata.javaClassName == "java.util.UUID" ->
+                typeOf<UUID>()
+
+            Types.REAL if tableColumnMetadata.javaClassName == "java.lang.Double" ->
+                typeOf<Double>()
+
+            Types.FLOAT if tableColumnMetadata.javaClassName == "java.lang.Double" ->
+                typeOf<Double>()
+
+            Types.NUMERIC if tableColumnMetadata.javaClassName == "java.lang.Double" ->
+                typeOf<Double>()
+
+            // Force BIGINT to always be Long, regardless of javaClassName
+            // Some JDBC drivers (e.g., MariaDB) may report Integer for small BIGINT values
+            // TODO: tableColumnMetadata.jdbcType == Types.BIGINT -> typeOf<Long>()
+
+            else if tableColumnMetadata.javaClassName == "[B" ->
+                typeOf<ByteArray>()
+
+            else if tableColumnMetadata.javaClassName == "java.sql.Blob" ->
+                typeOf<Blob>()
+
+            else ->
+                defaultJdbcTypeToKTypeMapping[tableColumnMetadata.jdbcType]
+                    ?: typeOf<String>()
+        }
+
+        return kType.withNullability(tableColumnMetadata.isNullable)
+    }
+
     /**
-     * Returns a [ColumnSchema] produced from [tableColumnMetadata].
+     * Extracts a value from the ResultSet for the given column.
+     * This method can be overridden by custom database types to change low-level reading logic.
+     *
+     * @param [rs] the ResultSet to read from
+     * @param [columnIndex] zero-based column index
+     * @return the extracted value, or null
      */
-    public abstract fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema?
+    public open fun <J : Any> getValueFromResultSet(
+        rs: ResultSet,
+        columnIndex: Int,
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+    ): J? =
+        try {
+            rs.getObject(columnIndex + 1)
+        } catch (_: Throwable) {
+            // TODO?
+            rs.getString(columnIndex + 1)
+        } as J?
+
+    // TODO add preprocessors for common types, like sql Arrays, Java datetimes, etc.
+    public open fun getPreprocessedValueType(
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+    ): KType =
+        when (tableColumnMetadata.jdbcType) {
+            Types.TIMESTAMP if tableColumnMetadata.javaClassName == "java.time.LocalDateTime" ->
+                typeOf<LocalDateTime>()
+
+            Types.TIMESTAMP ->
+                typeOf<Instant>()
+
+            Types.BINARY if tableColumnMetadata.javaClassName == "java.util.UUID" ->
+                typeOf<Uuid>()
+
+            else ->
+                expectedJdbcType
+        }.withNullability(tableColumnMetadata.isNullable)
+
+    // TODO add preprocessors for common types, like sql Arrays, Java datetimes, etc.
+    public open fun <J : Any, D : Any> preprocessValue(
+        value: J?,
+        tableColumnMetadata: TableColumnMetadata,
+        expectedJdbcType: KType,
+        expectedPreprocessedValueType: KType,
+    ): D? =
+        when (tableColumnMetadata.jdbcType) {
+            Types.TIMESTAMP if tableColumnMetadata.javaClassName == "java.time.LocalDateTime" ->
+                (value as JavaLocalDateTime?)?.toKotlinLocalDateTime()
+
+            Types.TIMESTAMP ->
+                (value as Timestamp?)?.toInstant()?.toKotlinInstant()
+
+            Types.BINARY if tableColumnMetadata.javaClassName == "java.util.UUID" ->
+                (value as UUID?)?.toKotlinUuid()
+
+            else ->
+                value
+        } as D?
+
+    public open fun getTargetColumnSchema(
+        tableColumnMetadata: TableColumnMetadata,
+        expectedValueType: KType,
+    ): ColumnSchema =
+        when (tableColumnMetadata.jdbcType) {
+            // buildDataColumn post-processes java.sql.Array -> Kotlin arrays, making the result type `Array<*>`
+            Types.ARRAY -> ColumnSchema.Value(typeOf<Array<*>>().withNullability(expectedValueType.isMarkedNullable))
+
+            else -> ColumnSchema.Value(expectedValueType)
+        }
+
+    public open fun <D : Any, P : Any> buildDataColumn(
+        name: String,
+        values: List<D?>,
+        tableColumnMetadata: TableColumnMetadata,
+        targetColumnSchema: ColumnSchema,
+        inferNullability: Boolean,
+    ): DataColumn<P?> {
+        val postProcessedValues = when (tableColumnMetadata.jdbcType) {
+            // Special case which post-processes java.sql.Array -> Kotlin arrays
+            Types.ARRAY -> handleArrayValues(values)
+
+            else -> values
+        }
+        return postProcessedValues.toDataColumn(
+            name = name,
+            targetColumnSchema = targetColumnSchema,
+            inferNullability = inferNullability,
+        )
+    }
+
+    protected fun <D : Any, P : Any> List<D?>.toDataColumn(
+        name: String,
+        targetColumnSchema: ColumnSchema,
+        inferNullability: Boolean,
+    ): DataColumn<P?> =
+        when (targetColumnSchema) {
+            is ColumnSchema.Value ->
+                DataColumn.createValueColumn(
+                    name = name,
+                    values = this,
+                    infer = if (inferNullability) Infer.Nulls else Infer.None,
+                    type = targetColumnSchema.type,
+                ).cast()
+
+            // NOTE: this case should be avoided.
+            //  Creating `n` DataRows is heavy!
+            is ColumnSchema.Group ->
+                DataColumn.createColumnGroup(
+                    name = name,
+                    df = (this as List<AnyRow>).toDataFrame(),
+                ).asDataColumn().cast()
+
+            is ColumnSchema.Frame ->
+                DataColumn.createFrameColumn(
+                    name = name,
+                    groups = this as List<AnyFrame>,
+                    schema = lazy { targetColumnSchema.schema },
+                ).cast()
+        }
 
     /**
      * Checks if the given table name is a system table for the specified database type.
@@ -103,14 +316,6 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
      */
     public abstract fun buildTableMetadata(tables: ResultSet): TableMetadata
 
-    /**
-     * Converts SQL data type to a Kotlin data type.
-     *
-     * @param [tableColumnMetadata] The metadata of the table column.
-     * @return The corresponding Kotlin data type, or null if no mapping is found.
-     */
-    public abstract fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType?
-
     /**
      * Builds a SELECT query for reading from a table.
      *
@@ -196,213 +401,6 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
         return connection
     }
 
-    /**
-     * Extracts a value from the ResultSet for the given column.
-     * This method can be overridden by custom database types to provide specialized parsing logic.
-     *
-     * @param [rs] the ResultSet to read from
-     * @param [columnIndex] zero-based column index
-     * @param [columnMetadata] metadata for the column
-     * @param [kType] the Kotlin type for this column
-     * @return the extracted value, or null
-     */
-    public open fun extractValueFromResultSet(
-        rs: ResultSet,
-        columnIndex: Int,
-        columnMetadata: TableColumnMetadata,
-        kType: KType,
-    ): Any? =
-        try {
-            rs.getObject(columnIndex + 1)
-            // TODO: add a special handler for Blob via Streams
-        } catch (_: Throwable) {
-            // TODO: expand for all the types like in generateKType function
-            if (kType.isSupertypeOf(String::class.starProjectedType)) {
-                rs.getString(columnIndex + 1)
-            } else {
-                rs.getString(columnIndex + 1)
-            }
-        }
-
-    /**
-     * Builds a single DataColumn with proper type handling.
-     * Accepts a mutable list to allow efficient post-processing.
-     */
-    public open fun buildDataColumn(
-        name: String,
-        values: MutableList<Any?>,
-        kType: KType,
-        inferNullability: Boolean,
-    ): DataColumn<*> {
-        val correctedValues = postProcessColumnValues(values, kType)
-
-        return DataColumn.createValueColumn(
-            name = name,
-            values = correctedValues,
-            infer = convertNullabilityInference(inferNullability),
-            type = kType,
-        )
-    }
-
-    private fun convertNullabilityInference(inferNullability: Boolean) =
-        if (inferNullability) Infer.Nulls else Infer.None
-
-    /**
-     * Processes the column values retrieved from the database and performs transformations based on the provided
-     * Kotlin type and column metadata. The method allows for custom post-processing logic, such as handling
-     * specific database column types, including arrays.
-     *
-     * @param values the list of raw values retrieved from the database for the column.
-     * @param kType the Kotlin type that the column values should be transformed to.
-     * @return a list of processed column values, with transformations applied where necessary, or the original list if no transformation is needed.
-     */
-    private fun postProcessColumnValues(values: MutableList<Any?>, kType: KType): List<Any?> =
-        when {
-            /* EXAMPLE: columnMetadata.sqlTypeName == "MY_CUSTOM_ARRAY" -> {
-                values.map { /* custom transformation */ }
-            } */
-            kType.classifier == Array::class -> {
-                handleArrayValues(values)
-            }
-
-            else -> values
-        }
-
-    /**
-     * Converts SQL Array objects to strongly-typed arrays.
-     *
-     * Extracts arrays from SQL Array objects and converts them to a consistent type
-     * if all elements share the same type. Returns original arrays if types vary.
-     *
-     * @param values raw values containing SQL Array objects
-     * @return list of consistently typed arrays, or original arrays if no common type exists
-     */
-    private fun handleArrayValues(values: MutableList<Any?>): List<Any> {
-        // Intermediate variable for the first mapping
-        val sqlArrays = values.mapNotNull {
-            (it as? java.sql.Array)?.array?.let { array -> array as? Array<*> }
-        }
-
-        // Flatten the arrays to iterate through all elements and filter out null values, then map to component types
-        val allElementTypes = sqlArrays
-            .flatMap { array ->
-                (array.javaClass.componentType?.kotlin?.let { listOf(it) } ?: emptyList())
-            } // Get the component type of each array and convert it to a Kotlin class, if available
-
-        // Find distinct types and ensure there's only one distinct type
-        val commonElementType = allElementTypes
-            .distinct() // Get unique element types
-            .singleOrNull() // Ensure there's only one unique element type, otherwise return null
-            ?: Any::class // Fallback to Any::class if multiple distinct types or no elements found
-
-        return if (commonElementType != Any::class) {
-            sqlArrays.map { castArray(it, commonElementType).toTypedArray() }
-        } else {
-            sqlArrays
-        }
-    }
-
-    /** Utility function to cast arrays based on the type of elements */
-    private fun <T : Any> castArray(array: Array<*>, elementType: KClass<T>): List<T> =
-        array.mapNotNull { elementType.safeCast(it) }
-
-    /**
-     * Creates a mapping between common SQL types and their corresponding KTypes.
-     *
-     * @param tableColumnMetadata The metadata of the table column.
-     * @return The KType associated with the SQL type or a default type if no mapping is found.
-     */
-    public open fun makeCommonSqlToKTypeMapping(tableColumnMetadata: TableColumnMetadata): KType {
-        val jdbcTypeToKTypeMapping = mapOf(
-            Types.BIT to Boolean::class,
-            Types.TINYINT to Int::class,
-            Types.SMALLINT to Int::class,
-            Types.INTEGER to Int::class,
-            Types.BIGINT to Long::class,
-            Types.FLOAT to Float::class,
-            Types.REAL to Float::class,
-            Types.DOUBLE to Double::class,
-            Types.NUMERIC to BigDecimal::class,
-            Types.DECIMAL to BigDecimal::class,
-            Types.CHAR to String::class,
-            Types.VARCHAR to String::class,
-            Types.LONGVARCHAR to String::class,
-            Types.DATE to Date::class,
-            Types.TIME to Time::class,
-            Types.TIMESTAMP to Timestamp::class,
-            Types.BINARY to ByteArray::class,
-            Types.VARBINARY to ByteArray::class,
-            Types.LONGVARBINARY to ByteArray::class,
-            Types.NULL to String::class,
-            Types.JAVA_OBJECT to Any::class,
-            Types.DISTINCT to Any::class,
-            Types.STRUCT to Any::class,
-            Types.ARRAY to Array::class,
-            Types.BLOB to ByteArray::class,
-            Types.CLOB to Clob::class,
-            Types.REF to Ref::class,
-            Types.DATALINK to Any::class,
-            Types.BOOLEAN to Boolean::class,
-            Types.ROWID to RowId::class,
-            Types.NCHAR to String::class,
-            Types.NVARCHAR to String::class,
-            Types.LONGNVARCHAR to String::class,
-            Types.NCLOB to NClob::class,
-            Types.SQLXML to SQLXML::class,
-            Types.REF_CURSOR to Ref::class,
-            Types.TIME_WITH_TIMEZONE to OffsetTime::class,
-            Types.TIMESTAMP_WITH_TIMEZONE to OffsetDateTime::class,
-        )
-
-        fun determineKotlinClass(tableColumnMetadata: TableColumnMetadata): KClass<*> =
-            when {
-                tableColumnMetadata.jdbcType == Types.OTHER -> when (tableColumnMetadata.javaClassName) {
-                    "[B" -> ByteArray::class
-                    else -> Any::class
-                }
-
-                tableColumnMetadata.javaClassName == "[B" -> ByteArray::class
-
-                tableColumnMetadata.javaClassName == "java.sql.Blob" -> Blob::class
-
-                tableColumnMetadata.jdbcType == Types.TIMESTAMP &&
-                    tableColumnMetadata.javaClassName == "java.time.LocalDateTime" -> LocalDateTime::class
-
-                tableColumnMetadata.jdbcType == Types.BINARY &&
-                    tableColumnMetadata.javaClassName == "java.util.UUID" -> UUID::class
-
-                tableColumnMetadata.jdbcType == Types.REAL &&
-                    tableColumnMetadata.javaClassName == "java.lang.Double" -> Double::class
-
-                tableColumnMetadata.jdbcType == Types.FLOAT &&
-                    tableColumnMetadata.javaClassName == "java.lang.Double" -> Double::class
-
-                tableColumnMetadata.jdbcType == Types.NUMERIC &&
-                    tableColumnMetadata.javaClassName == "java.lang.Double" -> Double::class
-
-                // Force BIGINT to always be Long, regardless of javaClassName
-                // Some JDBC drivers (e.g., MariaDB) may report Integer for small BIGINT values
-                // TODO: tableColumnMetadata.jdbcType == Types.BIGINT -> Long::class
-
-                else -> jdbcTypeToKTypeMapping[tableColumnMetadata.jdbcType] ?: String::class
-            }
-
-        fun createArrayTypeIfNeeded(kClass: KClass<*>, isNullable: Boolean): KType =
-            if (kClass == Array::class) {
-                val typeParam = kClass.typeParameters[0].createType()
-                kClass.createType(
-                    arguments = listOf(kotlin.reflect.KTypeProjection.invariant(typeParam)),
-                    nullable = isNullable,
-                )
-            } else {
-                kClass.createType(nullable = isNullable)
-            }
-
-        val kClass: KClass<*> = determineKotlinClass(tableColumnMetadata)
-        val kType = createArrayTypeIfNeeded(kClass, tableColumnMetadata.isNullable)
-        return kType
-    }
-
     /**
      * Retrieves column metadata from a JDBC ResultSet.
      *
@@ -415,7 +413,7 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
      * - `getTableName()` → extract from column name if contains '.' → `null`
      * - `isNullable()` → [DatabaseMetaData.getColumns] → `true` (assume nullable)
      * - `getColumnTypeName()` → `"OTHER"`
-     * - `getColumnType()` → [java.sql.Types.OTHER]
+     * - `getColumnType()` → [Types.OTHER]
      * - `getColumnDisplaySize()` → `0`
      * - `getColumnClassName()` → `"java.lang.Object"`
      *
@@ -522,12 +520,12 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
             val uniqueName = manageColumnNameDuplication(nameCounter, columnName)
 
             columns += TableColumnMetadata(
-                uniqueName,
-                columnType,
-                jdbcType,
-                displaySize,
-                javaClassName,
-                isNullable,
+                name = uniqueName,
+                sqlTypeName = columnType,
+                jdbcType = jdbcType,
+                size = displaySize,
+                javaClassName = javaClassName,
+                isNullable = isNullable,
             )
         }
 
@@ -558,4 +556,43 @@ public abstract class DbType(public val dbTypeInJdbcUrl: String) {
 
         return name
     }
+
+    /**
+     * todo?
+     * Converts SQL Array objects to strongly-typed arrays.
+     *
+     * Extracts arrays from SQL Array objects and converts them to a consistent type
+     * if all elements share the same type. Returns original arrays if types vary.
+     *
+     * @param values raw values containing SQL Array objects
+     * @return list of consistently typed arrays, or original arrays if no common type exists
+     */
+    private fun handleArrayValues(values: List<Any?>): List<Array<*>?> {
+        // Intermediate variable for the first mapping
+        val sqlArrays = values.mapNotNull {
+            (it as? java.sql.Array)?.array?.let { array -> array as? Array<*> }
+        }
+
+        // Flatten the arrays to iterate through all elements and filter out null values, then map to component types
+        val allElementTypes = sqlArrays
+            .flatMap { array ->
+                (array.javaClass.componentType?.kotlin?.let { listOf(it) } ?: emptyList())
+            } // Get the component type of each array and convert it to a Kotlin class, if available
+
+        // Find distinct types and ensure there's only one distinct type
+        val commonElementType = allElementTypes
+            .distinct() // Get unique element types
+            .singleOrNull() // Ensure there's only one unique element type, otherwise return null
+            ?: Any::class // Fallback to Any::class if multiple distinct types or no elements found
+
+        return if (commonElementType != Any::class) {
+            sqlArrays.map { castArray(it, commonElementType).toTypedArray() }
+        } else {
+            sqlArrays
+        }
+    }
+
+    /** Utility function to cast arrays based on the type of elements */
+    private fun <T : Any> castArray(array: Array<*>, elementType: KClass<T>): List<T> =
+        array.mapNotNull { elementType.safeCast(it) }
 }
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DuckDb.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DuckDb.kt
index d6882133e7..10b14ff4d8 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DuckDb.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/DuckDb.kt
@@ -1,7 +1,10 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
 import io.github.oshai.kotlinlogging.KotlinLogging
-import org.duckdb.DuckDBColumnType
+import kotlinx.datetime.LocalDate
+import kotlinx.datetime.LocalTime
+import kotlinx.datetime.toKotlinLocalDate
+import kotlinx.datetime.toKotlinLocalTime
 import org.duckdb.DuckDBColumnType.ARRAY
 import org.duckdb.DuckDBColumnType.BIGINT
 import org.duckdb.DuckDBColumnType.BIT
@@ -39,32 +42,43 @@ import org.duckdb.DuckDBColumnType.UUID
 import org.duckdb.DuckDBColumnType.VARCHAR
 import org.duckdb.DuckDBResultSetMetaData
 import org.duckdb.JsonNode
+import org.jetbrains.kotlinx.dataframe.AnyFrame
+import org.jetbrains.kotlinx.dataframe.AnyRow
 import org.jetbrains.kotlinx.dataframe.DataFrame
+import org.jetbrains.kotlinx.dataframe.DataRow
+import org.jetbrains.kotlinx.dataframe.api.asColumnGroup
+import org.jetbrains.kotlinx.dataframe.api.asDataColumn
+import org.jetbrains.kotlinx.dataframe.api.toDataFrame
+import org.jetbrains.kotlinx.dataframe.impl.schema.DataFrameSchemaImpl
 import org.jetbrains.kotlinx.dataframe.io.DbConnectionConfig
-import org.jetbrains.kotlinx.dataframe.io.db.DuckDb.convertSqlTypeToKType
 import org.jetbrains.kotlinx.dataframe.io.readAllSqlTables
 import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.math.BigDecimal
 import java.math.BigInteger
-import java.sql.Array
 import java.sql.Blob
 import java.sql.Connection
 import java.sql.DatabaseMetaData
 import java.sql.DriverManager
 import java.sql.ResultSet
 import java.sql.Struct
-import java.sql.Timestamp
-import java.time.LocalDate
-import java.time.LocalTime
-import java.time.OffsetDateTime
-import java.time.OffsetTime
 import java.util.Properties
-import java.util.UUID
-import kotlin.reflect.KType
+import kotlin.reflect.KClass
 import kotlin.reflect.KTypeProjection
 import kotlin.reflect.full.createType
+import kotlin.reflect.full.isSubtypeOf
 import kotlin.reflect.full.withNullability
 import kotlin.reflect.typeOf
+import kotlin.time.Instant
+import kotlin.time.toKotlinInstant
+import kotlin.uuid.Uuid
+import kotlin.uuid.toKotlinUuid
+import java.sql.Array as SqlArray
+import java.sql.Timestamp as SqlTimestamp
+import java.time.LocalDate as JavaLocalDate
+import java.time.LocalTime as JavaLocalTime
+import java.time.OffsetDateTime as JavaOffsetDateTime
+import java.time.OffsetTime as JavaOffsetTime
+import java.util.UUID as JavaUUID
 
 private val logger = KotlinLogging.logger {}
 
@@ -74,88 +88,211 @@ private val logger = KotlinLogging.logger {}
  * This class provides methods to convert data from a [ResultSet] to the appropriate type for DuckDB,
  * and to generate the corresponding [column schema][ColumnSchema].
  */
-public object DuckDb : DbType("duckdb") {
+public object DuckDb : AdvancedDbType("duckdb") {
 
     /** the name of the class of the DuckDB JDBC driver */
     override val driverClassName: String = "org.duckdb.DuckDBDriver"
 
+    override fun generateConverter(tableColumnMetadata: TableColumnMetadata): AnyJdbcToDataFrameConverter =
+        parseDuckDbType(tableColumnMetadata.sqlTypeName, tableColumnMetadata.isNullable)
+
+    private val duckDbTypeCache = mutableMapOf<Pair<String, Boolean>, AnyJdbcToDataFrameConverter>()
+
     /**
-     * How a column type from JDBC, [tableColumnMetadata], is read in Java/Kotlin.
+     * How a column type from JDBC, [sqlTypeName], is read in Java/Kotlin.
      * The returned type must exactly follow [ResultSet.getObject] of your specific database's JDBC driver.
      * Returning `null` defer the implementation to the default one (which may not always be correct).
      *
-     * Following [org.duckdb.DuckDBVector.getObject].
+     * Following [org.duckdb.DuckDBVector.getObject] and converting the result to
+     *
      */
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType =
-        tableColumnMetadata.sqlTypeName.toKType(tableColumnMetadata.isNullable)
+    internal fun parseDuckDbType(sqlTypeName: String, isNullable: Boolean): AnyJdbcToDataFrameConverter =
+        duckDbTypeCache.getOrPut(Pair(sqlTypeName, isNullable)) {
+            when (DuckDBResultSetMetaData.TypeNameToType(sqlTypeName)) {
+                BOOLEAN -> jdbcToDfConverterForValueColumnOf<Boolean>(isNullable)
 
-    /**
-     * How a column from JDBC should be represented as DataFrame (value) column
-     * See [convertSqlTypeToKType].
-     */
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema {
-        val type = convertSqlTypeToKType(tableColumnMetadata)
-        return ColumnSchema.Value(type)
-    }
+                TINYINT -> jdbcToDfConverterForValueColumnOf<Byte>(isNullable)
 
-    /**
-     * Follows exactly [org.duckdb.DuckDBVector.getObject].
-     *
-     * "// dataframe-jdbc" is added for all types that are covered correctly by
-     * [org.jetbrains.kotlinx.dataframe.io.db.DbType.makeCommonSqlToKTypeMapping] at the moment, however, to cover
-     * all nested types, we'll use a full type-map for all [DuckDB types][DuckDBColumnType] exactly.
-     */
-    @Suppress("ktlint:standard:blank-line-between-when-conditions")
-    internal fun String.toKType(isNullable: Boolean): KType {
-        val sqlTypeName = this
-        return when (DuckDBResultSetMetaData.TypeNameToType(sqlTypeName)) {
-            BOOLEAN -> typeOf<Boolean>() // dataframe-jdbc
-            TINYINT -> typeOf<Byte>()
-            SMALLINT -> typeOf<Short>()
-            INTEGER -> typeOf<Int>() // dataframe-jdbc
-            BIGINT -> typeOf<Long>() // dataframe-jdbc
-            HUGEINT -> typeOf<BigInteger>()
-            UHUGEINT -> typeOf<BigInteger>()
-            UTINYINT -> typeOf<Short>()
-            USMALLINT -> typeOf<Int>()
-            UINTEGER -> typeOf<Long>()
-            UBIGINT -> typeOf<BigInteger>()
-            FLOAT -> typeOf<Float>() // dataframe-jdbc
-            DOUBLE -> typeOf<Double>() // dataframe-jdbc
-            DECIMAL -> typeOf<BigDecimal>() // dataframe-jdbc
-            TIME -> typeOf<LocalTime>()
-            TIME_WITH_TIME_ZONE -> typeOf<OffsetTime>() // dataframe-jdbc
-            DATE -> typeOf<LocalDate>()
-            TIMESTAMP, TIMESTAMP_MS, TIMESTAMP_NS, TIMESTAMP_S -> typeOf<Timestamp>() // dataframe-jdbc
-            TIMESTAMP_WITH_TIME_ZONE -> typeOf<OffsetDateTime>() // dataframe-jdbc
-            JSON -> typeOf<JsonNode>()
-            BLOB -> typeOf<Blob>()
-            UUID -> typeOf<UUID>()
-            MAP -> {
-                val (key, value) = parseMapTypes(sqlTypeName)
-                Map::class.createType(
-                    listOf(
-                        KTypeProjection.invariant(key.toKType(false)),
-                        KTypeProjection.invariant(value.toKType(true)),
-                    ),
-                )
-            }
+                SMALLINT -> jdbcToDfConverterForValueColumnOf<Short>(isNullable)
+
+                INTEGER -> jdbcToDfConverterForValueColumnOf<Int>(isNullable)
+
+                BIGINT -> jdbcToDfConverterForValueColumnOf<Long>(isNullable)
+
+                HUGEINT -> jdbcToDfConverterForValueColumnOf<BigInteger>(isNullable)
+
+                UHUGEINT -> jdbcToDfConverterForValueColumnOf<BigInteger>(isNullable)
+
+                UTINYINT -> jdbcToDfConverterForValueColumnOf<Short>(isNullable)
+
+                USMALLINT -> jdbcToDfConverterForValueColumnOf<Int>(isNullable)
+
+                UINTEGER -> jdbcToDfConverterForValueColumnOf<Long>(isNullable)
+
+                UBIGINT -> jdbcToDfConverterForValueColumnOf<BigInteger>(isNullable)
+
+                FLOAT -> jdbcToDfConverterForValueColumnOf<Float>(isNullable)
+
+                DOUBLE -> jdbcToDfConverterForValueColumnOf<Double>(isNullable)
+
+                DECIMAL -> jdbcToDfConverterForValueColumnOf<BigDecimal>(isNullable)
+
+                TIME ->
+                    jdbcToDfConverterWithPreprocessingForValueColumnOf<JavaLocalTime, LocalTime>(isNullable) {
+                        it?.toKotlinLocalTime()
+                    }
+
+                // todo?
+                TIME_WITH_TIME_ZONE -> jdbcToDfConverterForValueColumnOf<JavaOffsetTime>(isNullable)
+
+                DATE -> jdbcToDfConverterWithPreprocessingForValueColumnOf<JavaLocalDate, LocalDate>(isNullable) {
+                    it?.toKotlinLocalDate()
+                }
+
+                TIMESTAMP, TIMESTAMP_MS, TIMESTAMP_NS, TIMESTAMP_S ->
+                    jdbcToDfConverterWithPreprocessingForValueColumnOf<SqlTimestamp, Instant>(isNullable) {
+                        it?.toInstant()?.toKotlinInstant()
+                    }
+
+                // todo?
+                TIMESTAMP_WITH_TIME_ZONE -> jdbcToDfConverterForValueColumnOf<JavaOffsetDateTime>(isNullable)
 
-            LIST, ARRAY -> {
-                // TODO requires #1266 and #1273 for specific types
-                //   val listType = parseListType(sqlTypeName)
-                //   Array::class.createType(
-                //       listOf(KTypeProjection.invariant(listType.toKType(true))),
-                //   )
-                typeOf<Array>()
+                // TODO!
+                JSON -> jdbcToDfConverterForValueColumnOf<JsonNode>(isNullable)
+
+                BLOB -> jdbcToDfConverterForValueColumnOf<Blob>(isNullable)
+
+                UUID -> jdbcToDfConverterWithPreprocessingForValueColumnOf<JavaUUID, Uuid>(isNullable) {
+                    it?.toKotlinUuid()
+                }
+
+                MAP -> {
+                    val (key, value) = parseMapTypes(sqlTypeName)
+
+                    val parsedKeyType = parseDuckDbType(key, false)
+                    val parsedValueType = parseDuckDbType(value, true).castToAny()
+
+                    val targetMapType = Map::class.createType(
+                        listOf(
+                            KTypeProjection.invariant(parsedKeyType.targetSchema.type),
+                            KTypeProjection.invariant(parsedValueType.targetSchema.type),
+                        ),
+                    ).withNullability(isNullable)
+
+                    jdbcToDfConverterWithPreprocessingForValueColumnOf<Map<String, Any?>, Map<String, Any?>>(
+                        isNullable = isNullable,
+                        preprocessedValueType = targetMapType,
+                    ) { map ->
+                        // only need to preprocess the values, as the keys are just Strings
+                        map?.mapValues { (_, value) ->
+                            parsedValueType.preprocessOrCast(value)
+                        }
+                    }
+                }
+
+                LIST, ARRAY -> {
+                    val listType = parseListType(sqlTypeName)
+                    val parsedListType =
+                        parseDuckDbType(listType, true).castToAny()
+
+                    val targetListType = List::class
+                        .createType(listOf(KTypeProjection.invariant(parsedListType.targetSchema.type)))
+                        .withNullability(isNullable)
+
+                    when (val listTargetSchema = parsedListType.targetSchema) {
+                        // convert STRUCT[] -> DataFrame<*> to create a FrameColumn
+                        is ColumnSchema.Group if parsedListType.expectedJdbcType.isSubtypeOf(typeOf<Struct?>()) ->
+                            jdbcToDfConverterWithPreprocessingFor<SqlArray, AnyFrame>(
+                                isNullable = isNullable,
+                                targetSchema = with(listTargetSchema) {
+                                    ColumnSchema.Frame(schema, nullable, contentType)
+                                },
+                            ) { sqlArray ->
+                                sqlArray
+                                    ?.toList()
+                                    ?.let { it as List<Struct?> }
+                                    ?.mapNotNull {
+                                        parsedListType.cast<Struct, Map<String, Any?>, AnyRow>()
+                                            .preprocessOrCast(it)
+                                    }?.toDataFrame()
+                            }
+
+                        else ->
+                            jdbcToDfConverterWithPreprocessingForValueColumnOf<SqlArray, List<Any?>>(
+                                isNullable = isNullable,
+                                preprocessedValueType = targetListType,
+                            ) { sqlArray ->
+                                sqlArray
+                                    ?.toList()
+                                    ?.map { parsedListType.preprocessOrCast(it) } // recursively preprocess
+                            }
+                    }
+                }
+
+                STRUCT -> {
+                    val structEntries = parseStructType(sqlTypeName)
+                    val parsedStructEntries = structEntries.mapValues { (_, type) ->
+                        parseDuckDbType(sqlTypeName = type, isNullable = true)
+                    }
+
+                    val targetSchema = ColumnSchema.Group(
+                        schema = DataFrameSchemaImpl(parsedStructEntries.mapValues { it.value.targetSchema }),
+                        contentType = typeOf<Any?>(),
+                    )
+
+                    jdbcToDfConverterWithProcessingFor<Struct, Map<String, Any?>, AnyRow>(
+                        isNullable = isNullable,
+                        targetSchema = targetSchema,
+                        valuePreprocessor = { struct ->
+                            // NOTE DataRows cannot be `null` in DataFrame, instead, all its fields become `null`
+                            if (struct == null) {
+                                parsedStructEntries.mapValues { null }
+                            } else {
+                                // read data from the struct
+                                val attrs = struct.getAttributes(
+                                    parsedStructEntries.mapValues {
+                                        (it.value.expectedJdbcType.classifier!! as KClass<*>).java
+                                    },
+                                )
+
+                                // and potentially, preprocess each value individually
+                                parsedStructEntries.entries.withIndex().associate { (i, entry) ->
+                                    entry.key to entry.value.castToAny().preprocessOrCast(attrs[i])
+                                }
+                            }
+                        },
+                        columnBuilder = { name, values, _ ->
+                            (values as List<Map<String, Any?>>)
+                                .toDataFrame()
+                                .asColumnGroup(name)
+                                .asDataColumn()
+                        },
+                    )
+                }
+
+                // Cannot handle this in Kotlin
+                UNION -> jdbcToDfConverterForValueColumnOf<Any>(isNullable)
+
+                VARCHAR -> jdbcToDfConverterForValueColumnOf<String>(isNullable)
+
+                UNKNOWN, BIT, INTERVAL, ENUM -> jdbcToDfConverterForValueColumnOf<String>(isNullable)
             }
+        }
 
-            STRUCT -> typeOf<Struct>() // TODO requires #1266 for specific types
-            UNION -> typeOf<Any>() // Cannot handle this in Kotlin
-            VARCHAR -> typeOf<String>()
-            UNKNOWN, BIT, INTERVAL, ENUM -> typeOf<String>()
-        }.withNullability(isNullable)
-    }
+    private fun SqlArray.toList(): List<Any?> =
+        when (val array = this.array) {
+            is IntArray -> array.toList()
+            is LongArray -> array.toList()
+            is ShortArray -> array.toList()
+            is ByteArray -> array.toList()
+            is FloatArray -> array.toList()
+            is DoubleArray -> array.toList()
+            is BooleanArray -> array.toList()
+            is CharArray -> array.toList()
+            is Array<*> -> array.toList()
+            is SqlArray -> array.toList()
+            else -> error("unknown array type $array")
+        }
 
     /** Parses "MAP(X, Y)" into "X" and "Y", taking parentheses into account */
     internal fun parseMapTypes(typeString: String): Pair<String, String> {
@@ -196,6 +333,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
@@ -228,7 +376,7 @@ public object DuckDb : DbType("duckdb") {
      * but supports read-only mode through connection parameters.
      *
      * @param [dbConfig] The database configuration containing URL, credentials, and read-only flag.
-     * @return A configured [java.sql.Connection] instance.
+     * @return A configured [Connection] instance.
      */
     override fun createConnection(dbConfig: DbConnectionConfig): Connection {
         val properties = Properties().apply {
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/H2.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/H2.kt
index 96cea43724..9eccf9bb1f 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/H2.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/H2.kt
@@ -1,6 +1,5 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
-import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.sql.ResultSet
 import java.util.Locale
 import kotlin.reflect.KType
@@ -119,8 +118,9 @@ public open class H2(public val mode: Mode = Mode.Regular) : DbType("h2") {
     override val driverClassName: String
         get() = "org.h2.Driver"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? =
-        delegate?.convertSqlTypeToColumnSchemaValue(tableColumnMetadata)
+    override fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType =
+        delegate?.getExpectedJdbcType(tableColumnMetadata)
+            ?: super.getExpectedJdbcType(tableColumnMetadata)
 
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean {
         val locale = Locale.getDefault()
@@ -146,9 +146,6 @@ public open class H2(public val mode: Mode = Mode.Regular) : DbType("h2") {
                 tables.getString("table_cat"),
             )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? =
-        delegate?.convertSqlTypeToKType(tableColumnMetadata)
-
     public override fun buildSqlQueryWithLimit(sqlQuery: String, limit: Int): String =
         delegate?.buildSqlQueryWithLimit(sqlQuery, limit) ?: super.buildSqlQueryWithLimit(sqlQuery, limit)
 }
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/JdbcToDataFrameConverter.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/JdbcToDataFrameConverter.kt
new file mode 100644
index 0000000000..3ebad6c0ee
--- /dev/null
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/JdbcToDataFrameConverter.kt
@@ -0,0 +1,291 @@
+package org.jetbrains.kotlinx.dataframe.io.db
+
+import org.jetbrains.kotlinx.dataframe.DataColumn
+import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
+import java.sql.ResultSet
+import kotlin.reflect.KType
+import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
+
+public typealias AnyJdbcToDataFrameConverter = JdbcToDataFrameConverter<*, *, *>
+
+/**
+ * Represents all type information that can be retrieved from an SQL column.
+ * This can be extended for your specific [DbType] if you need extra information.
+ *
+ * This class needs to be stateless, so it can be memoized in [AdvancedDbType.generateConverter].
+ *
+ *
+ * @param J the type of the value coming from the JDBC driver.
+ * @param D the type of the column values after preprocessing. Will be equal to [J] if [valuePreprocessor] is `null`.
+ * @param P the type of the column values after postprocessing. Will be equal to [D] if [columnBuilder] is `null`.
+ *
+ * @property targetSchema the target schema of the column after running the optional
+ *   [valuePreprocessor] and [columnBuilder].
+ * @property valuePreprocessor an optional function that converts values from [java.sql.ResultSet.getObject]
+ *   to a cell/row suitable to be put into a [DataColumn].
+ * @property columnBuilder an optional function that converts a [List] with values of type [D]
+ *   to a [DataColumn] of with values of type [P].
+ */
+public open class JdbcToDataFrameConverter<J : Any, D : Any, P : Any>(
+    public open val expectedJdbcType: KType,
+    public open val preprocessedValueType: KType,
+    public open val targetSchema: ColumnSchema,
+    public open val resultSetReader: DbResultSetReader<J>?,
+    public open val valuePreprocessor: DbValuePreprocessor<J, D>?,
+    public open val columnBuilder: DbColumnBuilder<D, P>?,
+) {
+    public open fun getValueFromResultSetOrElse(rs: ResultSet, columnIndex: Int, default: () -> J?): J? {
+        resultSetReader?.let { reader ->
+            return reader.getValue(rs, columnIndex)
+        }
+        return default()
+    }
+
+    public open fun preprocessOrCast(value: J?): D? {
+        valuePreprocessor?.let { valuePreprocessor ->
+            return valuePreprocessor.preprocess(value)
+        }
+        return value as D?
+    }
+
+    public open fun buildDataColumnOrNull(name: String, values: List<D?>, inferNullability: Boolean): DataColumn<P?>? {
+        columnBuilder?.let { columnPostprocessor ->
+            return columnPostprocessor.buildDataColumn(name, values, inferNullability)
+        }
+        return null
+    }
+}
+
+public fun <J : Any, D : Any, P : Any> JdbcToDataFrameConverter<*, *, *>.cast(): JdbcToDataFrameConverter<J, D, P> =
+    this as JdbcToDataFrameConverter<J, D, P>
+
+public fun JdbcToDataFrameConverter<*, *, *>.castToAny(): JdbcToDataFrameConverter<Any, Any, Any> = cast()
+
+// region generic constructors
+
+public fun <J : Any, D : Any, P : Any> jdbcToDfConverterWithProcessingFor(
+    jdbcSourceType: KType,
+    preprocessedValueType: KType, // = jdbcSourceType
+    targetSchema: ColumnSchema, // = ColumnSchema.Value(preprocessedValueType)
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+    columnBuilder: DbColumnBuilder<D, P>?,
+): JdbcToDataFrameConverter<J, D, P> =
+    JdbcToDataFrameConverter(
+        expectedJdbcType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+        columnBuilder = columnBuilder,
+    )
+
+public inline fun <reified J : Any, reified D : Any, P : Any> jdbcToDfConverterWithProcessingFor(
+    isNullable: Boolean,
+    jdbcSourceType: KType = typeOf<J>().withNullability(isNullable),
+    preprocessedValueType: KType = typeOf<D>().withNullability(isNullable),
+    targetSchema: ColumnSchema,
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+    columnBuilder: DbColumnBuilder<D, P>?,
+): JdbcToDataFrameConverter<J, D, P> =
+    jdbcToDfConverterWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+        columnBuilder = columnBuilder,
+    )
+
+public fun <J : Any> jdbcToDfConverterFor(
+    jdbcSourceType: KType,
+    preprocessedValueType: KType,
+    targetSchema: ColumnSchema,
+    resultSetReader: DbResultSetReader<J>? = null,
+): JdbcToDataFrameConverter<J, J, J> =
+    jdbcToDfConverterWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = null,
+        columnBuilder = null,
+    )
+
+public fun <J : Any, D : Any> jdbcToDfConverterWithPreprocessingFor(
+    jdbcSourceType: KType,
+    preprocessedValueType: KType,
+    targetSchema: ColumnSchema,
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+): JdbcToDataFrameConverter<J, D, D> =
+    jdbcToDfConverterWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+        columnBuilder = null,
+    )
+
+public inline fun <reified J : Any, reified D : Any> jdbcToDfConverterWithPreprocessingFor(
+    isNullable: Boolean,
+    jdbcSourceType: KType = typeOf<J>().withNullability(isNullable),
+    preprocessedValueType: KType = typeOf<D>().withNullability(isNullable),
+    targetSchema: ColumnSchema,
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+): JdbcToDataFrameConverter<J, D, D> =
+    jdbcToDfConverterWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+        columnBuilder = null,
+    )
+
+public fun <J : Any, P : Any> jdbcToDfConverterWithPostprocessingFor(
+    jdbcSourceType: KType,
+    targetSchema: ColumnSchema,
+    resultSetReader: DbResultSetReader<J>? = null,
+    columnBuilder: DbColumnBuilder<J, P>?,
+): JdbcToDataFrameConverter<J, J, P> =
+    jdbcToDfConverterWithProcessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = jdbcSourceType,
+        targetSchema = targetSchema,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = null,
+        columnBuilder = columnBuilder,
+    )
+
+// endregion
+
+// region ValueColumn constructors
+
+public fun <J : Any> jdbcToDfConverterForValueColumnOf(
+    kType: KType,
+    resultSetReader: DbResultSetReader<J>? = null,
+): JdbcToDataFrameConverter<J, J, J> =
+    jdbcToDfConverterFor(
+        jdbcSourceType = kType,
+        preprocessedValueType = kType,
+        targetSchema = ColumnSchema.Value(kType),
+        resultSetReader = resultSetReader,
+    )
+
+public inline fun <reified J : Any> jdbcToDfConverterForValueColumnOf(
+    isNullable: Boolean,
+    targetColumnType: KType = typeOf<J>().withNullability(isNullable),
+    resultSetReader: DbResultSetReader<J>? = null,
+): JdbcToDataFrameConverter<J, J, J> = jdbcToDfConverterForValueColumnOf(targetColumnType, resultSetReader)
+
+public fun <J : Any, D : Any> jdbcToDfConverterWithPreprocessingForValueColumnOf(
+    jdbcSourceType: KType,
+    preprocessedValueType: KType,
+    targetColumnType: KType,
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+): JdbcToDataFrameConverter<J, D, D> =
+    jdbcToDfConverterWithPreprocessingFor(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetSchema = ColumnSchema.Value(targetColumnType),
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+    )
+
+public inline fun <reified J : Any, reified D : Any> jdbcToDfConverterWithPreprocessingForValueColumnOf(
+    isNullable: Boolean,
+    jdbcSourceType: KType = typeOf<J>().withNullability(isNullable),
+    preprocessedValueType: KType = typeOf<D>().withNullability(isNullable),
+    targetColumnType: KType = preprocessedValueType,
+    resultSetReader: DbResultSetReader<J>? = null,
+    valuePreprocessor: DbValuePreprocessor<J, D>?,
+): JdbcToDataFrameConverter<J, D, D> =
+    jdbcToDfConverterWithPreprocessingForValueColumnOf(
+        jdbcSourceType = jdbcSourceType,
+        preprocessedValueType = preprocessedValueType,
+        targetColumnType = targetColumnType,
+        resultSetReader = resultSetReader,
+        valuePreprocessor = valuePreprocessor,
+    )
+
+public fun <J : Any, P : Any> jdbcToDfConverterWithPostprocessingForValueColumnOf(
+    jdbcSourceType: KType,
+    targetColumnType: KType,
+    resultSetReader: DbResultSetReader<J>? = null,
+    columnPostprocessor: DbColumnBuilder<J, P>?,
+): JdbcToDataFrameConverter<J, J, P> =
+    jdbcToDfConverterWithPostprocessingFor(
+        jdbcSourceType = jdbcSourceType,
+        targetSchema = ColumnSchema.Value(targetColumnType),
+        resultSetReader = resultSetReader,
+        columnBuilder = columnPostprocessor,
+    )
+
+public inline fun <reified J : Any, reified P : Any> jdbcToDfConverterWithPostprocessingForValueColumnOf(
+    isNullable: Boolean,
+    jdbcSourceType: KType = typeOf<J>().withNullability(isNullable),
+    targetColumnType: KType = typeOf<P>().withNullability(isNullable),
+    resultSetReader: DbResultSetReader<J>? = null,
+    columnPostprocessor: DbColumnBuilder<J, P>?,
+): JdbcToDataFrameConverter<J, J, P> =
+    jdbcToDfConverterWithPostprocessingForValueColumnOf(
+        jdbcSourceType = jdbcSourceType,
+        targetColumnType = targetColumnType,
+        resultSetReader = resultSetReader,
+        columnPostprocessor = columnPostprocessor,
+    )
+
+// endregion
+
+public fun interface DbResultSetReader<out J : Any> {
+
+    public fun getValue(rs: ResultSet, columnIndex: Int): J?
+}
+
+/**
+ * This preprocessor can be created for types where you want to convert the values
+ * coming from [java.sql.ResultSet.getObject] to a different type more suitable to be put in a [DataColumn]
+ *
+ * @param J the type of the value coming from the JDBC driver.
+ * @param D the type of the column values after preprocessing.
+ */
+public fun interface DbValuePreprocessor<in J : Any, out D : Any> {
+
+    /**
+     * Converts the given [jdbcValue]: [J] to a [D].
+     *
+     * If you intend to create a [org.jetbrains.kotlinx.dataframe.columns.ColumnGroup],
+     * return a [org.jetbrains.kotlinx.dataframe.DataRow] here.
+     *
+     * If you intend to create a [org.jetbrains.kotlinx.dataframe.columns.FrameColumn],
+     * return a [org.jetbrains.kotlinx.dataframe.DataFrame] here.
+     */
+    public fun preprocess(jdbcValue: J?): D?
+}
+
+public fun <J : Any, D : Any> DbValuePreprocessor<*, *>.cast(): DbValuePreprocessor<J, D> =
+    this as DbValuePreprocessor<J, D>
+
+public fun DbValuePreprocessor<*, *>.castToAny(): DbValuePreprocessor<Any, Any> = cast()
+
+/**
+ * @param D the type of the column values before entering the column.
+ * @param P the type of the column values after entering the column.
+ */
+public fun interface DbColumnBuilder<in D : Any, out P : Any> {
+
+    /**
+     * Converts the given [values]: [DataColumn] with values of type [D] to a [DataColumn] of with values of type [P].
+     */
+    public fun buildDataColumn(name: String, values: List<D?>, inferNullability: Boolean): DataColumn<P?>
+}
+
+public fun <D : Any, P : Any> DbColumnBuilder<*, *>.cast(): DbColumnBuilder<D, P> = this as DbColumnBuilder<D, P>
+
+public fun DbColumnBuilder<*, *>.castToAny(): DbColumnBuilder<Any, Any> = cast()
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MariaDb.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MariaDb.kt
index d91b60a0b2..9060004e62 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MariaDb.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MariaDb.kt
@@ -1,11 +1,9 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
-import org.jetbrains.kotlinx.dataframe.io.db.TableColumnMetadata
-import org.jetbrains.kotlinx.dataframe.io.db.TableMetadata
-import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.sql.ResultSet
 import kotlin.reflect.KType
-import kotlin.reflect.full.createType
+import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
 
 /**
  * Represents the MariaDb database type.
@@ -17,7 +15,7 @@ public object MariaDb : DbType("mariadb") {
     override val driverClassName: String
         get() = "org.mariadb.jdbc.Driver"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? {
+    override fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType {
         // Force BIGINT to always be Long, regardless of javaClassName
         // MariaDB JDBC driver may report Integer for small BIGINT values
         // TODO: investigate the corner case
@@ -30,15 +28,13 @@ public object MariaDb : DbType("mariadb") {
         if (tableColumnMetadata.sqlTypeName == "INTEGER UNSIGNED" ||
             tableColumnMetadata.sqlTypeName == "INT UNSIGNED"
         ) {
-            val kType = Long::class.createType(nullable = tableColumnMetadata.isNullable)
-            return ColumnSchema.Value(kType)
+            return typeOf<Long>().withNullability(tableColumnMetadata.isNullable)
         }
 
         if (tableColumnMetadata.sqlTypeName == "SMALLINT" && tableColumnMetadata.javaClassName == "java.lang.Short") {
-            val kType = Short::class.createType(nullable = tableColumnMetadata.isNullable)
-            return ColumnSchema.Value(kType)
+            return typeOf<Short>().withNullability(tableColumnMetadata.isNullable)
         }
-        return null
+        return super.getExpectedJdbcType(tableColumnMetadata)
     }
 
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean = MySql.isSystemTable(tableMetadata)
@@ -50,26 +46,6 @@ public object MariaDb : DbType("mariadb") {
             tables.getString("table_cat"),
         )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? {
-        // Force BIGINT to always be Long, regardless of javaClassName
-        // MariaDB JDBC driver may report Integer for small BIGINT values
-        // TODO: investigate the corner case
-        //        if (tableColumnMetadata.jdbcType == java.sql.Types.BIGINT) {
-        //            return Long::class.createType(nullable = tableColumnMetadata.isNullable)
-        //        }
-
-        if (tableColumnMetadata.sqlTypeName == "INTEGER UNSIGNED" ||
-            tableColumnMetadata.sqlTypeName == "INT UNSIGNED"
-        ) {
-            return Long::class.createType(nullable = tableColumnMetadata.isNullable)
-        }
-
-        if (tableColumnMetadata.sqlTypeName == "SMALLINT" && tableColumnMetadata.javaClassName == "java.lang.Short") {
-            return Short::class.createType(nullable = tableColumnMetadata.isNullable)
-        }
-        return null
-    }
-
     override fun quoteIdentifier(name: String): String {
         // schema.table -> `schema`.`table`
         return name.split(".").joinToString(".") { "`$it`" }
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MsSql.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MsSql.kt
index 2709e04b6a..78f3c9c5a7 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MsSql.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MsSql.kt
@@ -17,8 +17,6 @@ public object MsSql : DbType("sqlserver") {
     override val driverClassName: String
         get() = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? = null
-
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean {
         val locale = Locale.getDefault()
 
@@ -47,8 +45,6 @@ public object MsSql : DbType("sqlserver") {
             tables.getString("table_cat"),
         )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? = null
-
     public override fun buildSqlQueryWithLimit(sqlQuery: String, limit: Int): String =
         sqlQuery.replace("SELECT", "SELECT TOP $limit", ignoreCase = true)
 
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MySql.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MySql.kt
index e411345879..7c4f74486c 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MySql.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/MySql.kt
@@ -1,12 +1,10 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
-import org.jetbrains.kotlinx.dataframe.io.db.TableColumnMetadata
-import org.jetbrains.kotlinx.dataframe.io.db.TableMetadata
-import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.sql.ResultSet
 import java.util.Locale
 import kotlin.reflect.KType
-import kotlin.reflect.full.createType
+import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
 
 /**
  * Represents the MySql database type.
@@ -18,12 +16,11 @@ public object MySql : DbType("mysql") {
     override val driverClassName: String
         get() = "com.mysql.jdbc.Driver"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? {
+    override fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType {
         if (tableColumnMetadata.sqlTypeName == "INT UNSIGNED") {
-            val kType = Long::class.createType(nullable = tableColumnMetadata.isNullable)
-            return ColumnSchema.Value(kType)
+            return typeOf<Long>().withNullability(tableColumnMetadata.isNullable)
         }
-        return null
+        return super.getExpectedJdbcType(tableColumnMetadata)
     }
 
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean {
@@ -49,13 +46,6 @@ public object MySql : DbType("mysql") {
             tables.getString("table_cat"),
         )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? {
-        if (tableColumnMetadata.sqlTypeName == "INT UNSIGNED") {
-            return Long::class.createType(nullable = tableColumnMetadata.isNullable)
-        }
-        return null
-    }
-
     override fun quoteIdentifier(name: String): String {
         // schema.table -> `schema`.`table`
         return name.split(".").joinToString(".") { "`$it`" }
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/PostgreSql.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/PostgreSql.kt
index 8da1a66833..581c36d669 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/PostgreSql.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/PostgreSql.kt
@@ -1,12 +1,10 @@
 package org.jetbrains.kotlinx.dataframe.io.db
 
-import org.jetbrains.kotlinx.dataframe.io.db.TableColumnMetadata
-import org.jetbrains.kotlinx.dataframe.io.db.TableMetadata
-import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.sql.ResultSet
 import java.util.Locale
 import kotlin.reflect.KType
-import kotlin.reflect.full.createType
+import kotlin.reflect.full.withNullability
+import kotlin.reflect.typeOf
 
 /**
  * Represents the PostgreSql database type.
@@ -18,14 +16,14 @@ public object PostgreSql : DbType("postgresql") {
     override val driverClassName: String
         get() = "org.postgresql.Driver"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? {
-        // TODO: could be a wrapper of convertSqlTypeToKType
+    override fun getExpectedJdbcType(tableColumnMetadata: TableColumnMetadata): KType {
         // because of https://github.com/pgjdbc/pgjdbc/issues/425
         if (tableColumnMetadata.sqlTypeName == "money") {
-            val kType = String::class.createType(nullable = tableColumnMetadata.isNullable)
-            return ColumnSchema.Value(kType)
+            return typeOf<String>().withNullability(tableColumnMetadata.isNullable)
         }
-        return null
+        // TODO: Composite types like tableColumnMetadata.sqlTypeName = ROW("a" INTEGER, "b" CHARACTER VARYING(10))
+
+        return super.getExpectedJdbcType(tableColumnMetadata)
     }
 
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean =
@@ -39,15 +37,6 @@ public object PostgreSql : DbType("postgresql") {
             tables.getString("table_cat"),
         )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? {
-        // because of https://github.com/pgjdbc/pgjdbc/issues/425
-        if (tableColumnMetadata.sqlTypeName == "money") {
-            return String::class.createType(nullable = tableColumnMetadata.isNullable)
-        }
-
-        return null
-    }
-
     override fun quoteIdentifier(name: String): String {
         // schema.table -> "schema"."table"
         return name.split(".").joinToString(".") { "\"$it\"" }
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/Sqlite.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/Sqlite.kt
index adb03b2753..1073e12be5 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/Sqlite.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/db/Sqlite.kt
@@ -20,8 +20,6 @@ public object Sqlite : DbType("sqlite") {
     override val driverClassName: String
         get() = "org.sqlite.JDBC"
 
-    override fun convertSqlTypeToColumnSchemaValue(tableColumnMetadata: TableColumnMetadata): ColumnSchema? = null
-
     override fun isSystemTable(tableMetadata: TableMetadata): Boolean = tableMetadata.name.startsWith("sqlite_")
 
     override fun buildTableMetadata(tables: ResultSet): TableMetadata =
@@ -31,8 +29,6 @@ public object Sqlite : DbType("sqlite") {
             tables.getString("TABLE_CAT"),
         )
 
-    override fun convertSqlTypeToKType(tableColumnMetadata: TableColumnMetadata): KType? = null
-
     override fun createConnection(dbConfig: DbConnectionConfig): Connection =
         if (dbConfig.readOnly) {
             val config = SQLiteConfig()
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readDataFrameSchema.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readDataFrameSchema.kt
index acce35f9fc..8b07bf364c 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readDataFrameSchema.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readDataFrameSchema.kt
@@ -4,15 +4,12 @@ import org.jetbrains.kotlinx.dataframe.DataFrame
 import org.jetbrains.kotlinx.dataframe.api.schema
 import org.jetbrains.kotlinx.dataframe.impl.schema.DataFrameSchemaImpl
 import org.jetbrains.kotlinx.dataframe.io.db.DbType
-import org.jetbrains.kotlinx.dataframe.io.db.TableColumnMetadata
 import org.jetbrains.kotlinx.dataframe.io.db.extractDBTypeFromConnection
-import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import org.jetbrains.kotlinx.dataframe.schema.DataFrameSchema
 import java.sql.Connection
 import java.sql.DriverManager
 import java.sql.ResultSet
 import javax.sql.DataSource
-import kotlin.use
 
 /**
  * Retrieves the schema for an SQL table using the provided database configuration.
@@ -98,6 +95,8 @@ public fun DataFrameSchema.Companion.readSqlTable(
 ): DataFrameSchema {
     val determinedDbType = dbType ?: extractDBTypeFromConnection(connection)
 
+    // TODO don't need to read 1 row, take it just from TableColumnMetadatas
+
     // Read just 1 row to get the schema
     val singleRowDataFrame = DataFrame.readSqlTable(
         connection = connection,
@@ -322,7 +321,10 @@ public fun Connection.readDataFrameSchema(sqlQueryOrTableName: String, dbType: D
  */
 public fun DataFrameSchema.Companion.readResultSet(resultSet: ResultSet, dbType: DbType): DataFrameSchema {
     val tableColumns = getTableColumnsMetadata(resultSet, dbType)
-    return buildSchemaByTableColumns(tableColumns, dbType)
+    val expectedJdbcTypes = getExpectedJdbcTypes(dbType, tableColumns)
+    val preprocessedValueTypes = getPreprocessedValueTypes(dbType, tableColumns, expectedJdbcTypes)
+    val targetColumnSchemas = getTargetColumnSchemas(dbType, tableColumns, preprocessedValueTypes)
+    return DataFrameSchemaImpl(targetColumnSchemas)
 }
 
 /**
@@ -437,28 +439,3 @@ public fun DataFrameSchema.Companion.readAllSqlTables(
 
     return dataFrameSchemas
 }
-
-/**
- * Builds a DataFrame schema based on the given table columns.
- *
- * @param [tableColumns] a mutable map containing the table columns, where the key represents the column name
- * and the value represents the metadata of the column
- * @param [dbType] the type of database.
- * @return a [DataFrameSchema] object representing the schema built from the table columns.
- */
-internal fun buildSchemaByTableColumns(
-    tableColumns: MutableList<TableColumnMetadata>,
-    dbType: DbType,
-): DataFrameSchema {
-    val schemaColumns = tableColumns.associate {
-        Pair(it.name, generateColumnSchemaValue(dbType, it))
-    }
-
-    return DataFrameSchemaImpl(
-        columns = schemaColumns,
-    )
-}
-
-internal fun generateColumnSchemaValue(dbType: DbType, tableColumnMetadata: TableColumnMetadata): ColumnSchema =
-    dbType.convertSqlTypeToColumnSchemaValue(tableColumnMetadata)
-        ?: ColumnSchema.Value(dbType.makeCommonSqlToKTypeMapping(tableColumnMetadata))
diff --git a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readJdbc.kt b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readJdbc.kt
index b565ad3d5b..ddbabb7e6d 100644
--- a/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readJdbc.kt
+++ b/dataframe-jdbc/src/main/kotlin/org/jetbrains/kotlinx/dataframe/io/readJdbc.kt
@@ -1,19 +1,27 @@
 package org.jetbrains.kotlinx.dataframe.io
 
 import io.github.oshai.kotlinlogging.KotlinLogging
+import org.jetbrains.kotlinx.dataframe.AnyCol
 import org.jetbrains.kotlinx.dataframe.AnyFrame
 import org.jetbrains.kotlinx.dataframe.DataFrame
+import org.jetbrains.kotlinx.dataframe.api.isColumnGroup
+import org.jetbrains.kotlinx.dataframe.api.isFrameColumn
+import org.jetbrains.kotlinx.dataframe.api.isValueColumn
+import org.jetbrains.kotlinx.dataframe.api.schema
 import org.jetbrains.kotlinx.dataframe.api.toDataFrame
 import org.jetbrains.kotlinx.dataframe.io.db.DbType
 import org.jetbrains.kotlinx.dataframe.io.db.TableColumnMetadata
 import org.jetbrains.kotlinx.dataframe.io.db.extractDBTypeFromConnection
+import org.jetbrains.kotlinx.dataframe.schema.ColumnSchema
 import java.sql.Connection
 import java.sql.DatabaseMetaData
 import java.sql.DriverManager
 import java.sql.PreparedStatement
 import java.sql.ResultSet
 import javax.sql.DataSource
+import kotlin.reflect.KClass
 import kotlin.reflect.KType
+import kotlin.reflect.full.isSubclassOf
 
 private val logger = KotlinLogging.logger {}
 
@@ -181,7 +189,7 @@ private fun executeQueryAndBuildDataFrame(
             logger.debug { "Executing query: $sqlQuery" }
             statement.executeQuery().use { rs ->
                 val tableColumns = getTableColumnsMetadata(rs, determinedDbType)
-                fetchAndConvertDataFromResultSet(tableColumns, rs, determinedDbType, limit, inferNullability)
+                fetchAndConvertDataFromResultSet(determinedDbType, tableColumns, rs, limit, inferNullability)
             }
         }
     } catch (e: java.sql.SQLException) {
@@ -562,7 +570,7 @@ public fun DataFrame.Companion.readResultSet(
 ): AnyFrame {
     validateLimit(limit)
     val tableColumns = getTableColumnsMetadata(resultSet, dbType)
-    return fetchAndConvertDataFromResultSet(tableColumns, resultSet, dbType, limit, inferNullability)
+    return fetchAndConvertDataFromResultSet(dbType, tableColumns, resultSet, limit, inferNullability)
 }
 
 /**
@@ -851,8 +859,8 @@ private fun readTableAsDataFrame(
     return dataFrame
 }
 
-internal fun getTableColumnsMetadata(resultSet: ResultSet, dbType: DbType): MutableList<TableColumnMetadata> =
-    dbType.getTableColumnsMetadata(resultSet).toMutableList()
+internal fun getTableColumnsMetadata(resultSet: ResultSet, dbType: DbType): List<TableColumnMetadata> =
+    dbType.getTableColumnsMetadata(resultSet)
 
 /**
  * Fetches and converts data from a ResultSet into a mutable map.
@@ -866,15 +874,43 @@ internal fun getTableColumnsMetadata(resultSet: ResultSet, dbType: DbType): Muta
  * @return A mutable map containing the fetched and converted data.
  */
 internal fun fetchAndConvertDataFromResultSet(
-    tableColumns: MutableList<TableColumnMetadata>,
-    rs: ResultSet,
     dbType: DbType,
+    tableColumns: List<TableColumnMetadata>,
+    rs: ResultSet,
     limit: Int?,
     inferNullability: Boolean,
 ): AnyFrame {
-    val columnKTypes = buildColumnKTypes(tableColumns, dbType)
-    val columnData = readAllRowsFromResultSet(rs, tableColumns, columnKTypes, dbType, limit)
-    val dataFrame = buildDataFrameFromColumnData(columnData, tableColumns, columnKTypes, dbType, inferNullability)
+    val expectedJdbcTypes = getExpectedJdbcTypes(
+        dbType = dbType,
+        tableColumns = tableColumns,
+    )
+    val preprocessedValueTypes = getPreprocessedValueTypes(
+        dbType = dbType,
+        tableColumns = tableColumns,
+        expectedJdbcTypes = expectedJdbcTypes,
+    )
+    val targetColumnSchemas = getTargetColumnSchemas(
+        dbType = dbType,
+        tableColumns = tableColumns,
+        preprocessedValueTypes = preprocessedValueTypes,
+    )
+
+    val columnData = readAndPreprocessRowsFromResultSet(
+        rs = rs,
+        tableColumns = tableColumns,
+        expectedJdbcTypes = expectedJdbcTypes,
+        preprocessedValueTypes = preprocessedValueTypes,
+        dbType = dbType,
+        limit = limit,
+    )
+
+    val dataFrame = buildDataFrameFromColumnData(
+        dbType = dbType,
+        columnData = columnData,
+        tableColumns = tableColumns,
+        targetColumnSchemas = targetColumnSchemas,
+        inferNullability = inferNullability,
+    )
 
     logger.debug {
         "DataFrame with ${dataFrame.rowsCount()} rows and ${dataFrame.columnsCount()} columns created as a result of SQL query."
@@ -883,38 +919,69 @@ internal fun fetchAndConvertDataFromResultSet(
     return dataFrame
 }
 
-/**
- * Builds a map of column indices to their Kotlin types.
- */
-private fun buildColumnKTypes(tableColumns: List<TableColumnMetadata>, dbType: DbType): Map<Int, KType> =
-    tableColumns.indices.associateWith { index ->
-        generateKType(dbType, tableColumns[index])
+internal fun getExpectedJdbcTypes(dbType: DbType, tableColumns: List<TableColumnMetadata>): Map<String, KType> =
+    tableColumns.associate {
+        it.name to dbType.getExpectedJdbcType(tableColumnMetadata = it)
+    }
+
+internal fun getPreprocessedValueTypes(
+    dbType: DbType,
+    tableColumns: List<TableColumnMetadata>,
+    expectedJdbcTypes: Map<String, KType>,
+): Map<String, KType> =
+    tableColumns.associate {
+        it.name to dbType.getPreprocessedValueType(
+            tableColumnMetadata = it,
+            expectedJdbcType = expectedJdbcTypes[it.name]!!,
+        )
+    }
+
+internal fun getTargetColumnSchemas(
+    dbType: DbType,
+    tableColumns: List<TableColumnMetadata>,
+    preprocessedValueTypes: Map<String, KType>,
+): Map<String, ColumnSchema> =
+    tableColumns.associate {
+        it.name to dbType.getTargetColumnSchema(
+            tableColumnMetadata = it,
+            expectedValueType = preprocessedValueTypes[it.name]!!,
+        )
     }
 
 /**
  * Reads all rows from ResultSet and returns a column-oriented data structure.
- * Returns mutable lists to allow efficient post-processing without copying.
  */
-private fun readAllRowsFromResultSet(
+private fun readAndPreprocessRowsFromResultSet(
+    dbType: DbType,
     rs: ResultSet,
     tableColumns: List<TableColumnMetadata>,
-    columnKTypes: Map<Int, KType>,
-    dbType: DbType,
+    expectedJdbcTypes: Map<String, KType>,
+    preprocessedValueTypes: Map<String, KType>,
     limit: Int?,
-): List<MutableList<Any?>> {
-    val columnsCount = tableColumns.size
-    val columnData = List(columnsCount) { mutableListOf<Any?>() }
+): Map<String, List<Any?>> {
+    val columnNames = tableColumns.map { it.name }
+    val columnData = columnNames.associateWith { mutableListOf<Any?>() }
     var rowsRead = 0
 
     while (rs.next() && (limit == null || rowsRead < limit)) {
-        repeat(columnsCount) { columnIndex ->
-            val value = dbType.extractValueFromResultSet(
+        columnNames.forEachIndexed { i, name ->
+            val tableColumnMetadata = tableColumns[i]
+            val expectedJdbcType = expectedJdbcTypes[name]!!
+            val preprocessedValueType = preprocessedValueTypes[name]!!
+
+            val value = dbType.getValueFromResultSet<Any>(
                 rs = rs,
-                columnIndex = columnIndex,
-                columnMetadata = tableColumns[columnIndex],
-                kType = columnKTypes.getValue(columnIndex),
+                columnIndex = i,
+                tableColumnMetadata = tableColumnMetadata,
+                expectedJdbcType = expectedJdbcType,
+            )
+            val preprocessedValue = dbType.preprocessValue<Any, Any>(
+                value = value,
+                tableColumnMetadata = tableColumnMetadata,
+                expectedJdbcType = expectedJdbcType,
+                expectedPreprocessedValueType = preprocessedValueType,
             )
-            columnData[columnIndex].add(value)
+            columnData[name]!!.add(preprocessedValue)
         }
         rowsRead++
         // if (rowsRead % 1000 == 0) logger.debug { "Loaded $rowsRead rows." } // TODO: https://github.com/Kotlin/dataframe/issues/455
@@ -928,29 +995,88 @@ private fun readAllRowsFromResultSet(
  * Accepts mutable lists to enable efficient in-place transformations.
  */
 private fun buildDataFrameFromColumnData(
-    columnData: List<MutableList<Any?>>,
-    tableColumns: List<TableColumnMetadata>,
-    columnKTypes: Map<Int, KType>,
     dbType: DbType,
+    columnData: Map<String, List<Any?>>,
+    tableColumns: List<TableColumnMetadata>,
+    targetColumnSchemas: Map<String, ColumnSchema>,
     inferNullability: Boolean,
+    checkSchema: Boolean = true, // TODO add as configurable parameter
 ): AnyFrame =
-    columnData.mapIndexed { index, values ->
-        dbType.buildDataColumn(
-            name = tableColumns[index].name,
-            values = values,
-            kType = columnKTypes.getValue(index),
+    tableColumns.map {
+        val name = it.name
+        val column = dbType.buildDataColumn<Any, Any>(
+            name = name,
+            values = columnData[name]!!,
+            tableColumnMetadata = it,
+            targetColumnSchema = targetColumnSchemas[name]!!,
             inferNullability = inferNullability,
         )
+
+        if (checkSchema) {
+            column.checkSchema(targetColumnSchemas[name]!!)
+        }
+
+        column
     }.toDataFrame()
 
-/**
- * Generates a KType based on the given database type and table column metadata.
- *
- * @param dbType The database type.
- * @param tableColumnMetadata The table column metadata.
- *
- * @return The generated KType.
- */
-internal fun generateKType(dbType: DbType, tableColumnMetadata: TableColumnMetadata): KType =
-    dbType.convertSqlTypeToKType(tableColumnMetadata)
-        ?: dbType.makeCommonSqlToKTypeMapping(tableColumnMetadata)
+private fun AnyCol.checkSchema(expected: ColumnSchema) {
+    when (expected) {
+        is ColumnSchema.Value -> {
+            require(this.isValueColumn()) {
+                """
+                Found mismatching schema for column '${this.name()}'.
+                Column ${this.name()} is expected to be a value column of type ${expected.type} but it is ${this.type()}.
+                """.trimIndent()
+            }
+            require(values().all { it == null || it::class.isSubclassOf(expected.type.classifier as KClass<*>) }) {
+                """
+                Found mismatching type for value column '${this.name()}'.
+                Expected type: ${expected.type}
+                Actual types: ${values().map { it?.javaClass?.name ?: "null" }.distinct()}
+                """.trimIndent()
+            }
+        }
+
+        is ColumnSchema.Group -> {
+            require(this.isColumnGroup()) {
+                """
+                Found mismatching schema for column '${name()}'.
+                Column ${this.name()} is expected to be a column group but it is ${this.type()}.
+                """.trimIndent()
+            }
+            require(expected.schema.compare(this.schema()).isSuperOrMatches()) {
+                """
+                Found mismatching schema for column group '${name()}'.
+                Expected schema:
+                ${expected.schema}
+                
+                Actual schema:
+                ${this.schema()}
+                """.trimIndent()
+            }
+        }
+
+        is ColumnSchema.Frame -> {
+            require(this.isFrameColumn()) {
+                """
+                Found mismatching schema for column '${this.name()}'.
+                Column ${this.name()} is expected to be a frame column but it is ${this.type()}.
+                """.trimIndent()
+            }
+            require(values().all { expected.schema.compare(it.schema()).isSuperOrMatches() }) {
+                """
+                Found mismatching schema for frame column '${this.name()}'.
+                Expected schema:
+                ${expected.schema}
+                
+                Actual (deviating) schemas:
+                ${
+                    values().map { it.schema() }
+                        .distinct()
+                        .filterNot { expected.schema.compare(it).isSuperOrMatches() }
+                }
+                """.trimIndent()
+            }
+        }
+    }
+}
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/commonTestScenarios.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/commonTestScenarios.kt
index 2669f7e944..39c55f0ab7 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/commonTestScenarios.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/commonTestScenarios.kt
@@ -1,5 +1,11 @@
 package org.jetbrains.kotlinx.dataframe.io
 
+import io.kotest.assertions.Actual
+import io.kotest.assertions.AssertionFailedError
+import io.kotest.assertions.Exceptions
+import io.kotest.assertions.Expected
+import io.kotest.assertions.failure
+import io.kotest.assertions.print.printed
 import io.kotest.assertions.withClue
 import io.kotest.matchers.shouldBe
 import org.intellij.lang.annotations.Language
@@ -144,17 +150,11 @@ internal fun inferNullability(connection: Connection) {
  */
 @Suppress("INVISIBLE_REFERENCE")
 fun AnyFrame.assertInferredTypesMatchSchema() {
-    withClue({
-        """
-        |Inferred schema must be <: Provided schema
-        |
-        |Inferred Schema: 
-        |${inferType().schema().toString().lines().joinToString("\n|")}
-        |
-        |Provided Schema:
-        |${schema().toString().lines().joinToString("\n|")}
-        """.trimMargin()
-    }) {
-        schema().compare(inferType().schema()).isSuperOrMatches() shouldBe true
+    if (!schema().compare(inferType().schema()).isSuperOrMatches()) {
+        throw failure(
+            expected = Expected(inferType().schema().toString().lines().sorted().joinToString("\n").printed()),
+            actual = Actual(schema().toString().lines().sorted().joinToString("\n").printed()),
+            prependMessage = "Inferred schema must be <: Provided schema",
+        )
     }
 }
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mariadbH2Test.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mariadbH2Test.kt
index 0707000ef6..b4c99468f5 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mariadbH2Test.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mariadbH2Test.kt
@@ -21,8 +21,10 @@ import java.sql.Blob
 import java.sql.Connection
 import java.sql.DriverManager
 import java.sql.SQLException
+import java.sql.Timestamp
 import java.util.Date
 import kotlin.reflect.typeOf
+import kotlin.time.Instant
 
 private const val URL = "jdbc:h2:mem:test1;DB_CLOSE_DELAY=-1;MODE=MariaDB;DATABASE_TO_LOWER=TRUE"
 
@@ -230,8 +232,8 @@ class MariadbH2Test {
                     st.setDouble(11, i * 10.0)
                     st.setBigDecimal(12, BigDecimal(i * 10))
                     st.setDate(13, java.sql.Date(System.currentTimeMillis()))
-                    st.setTimestamp(14, java.sql.Timestamp(System.currentTimeMillis()))
-                    st.setTimestamp(15, java.sql.Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(14, Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(15, Timestamp(System.currentTimeMillis()))
                     st.setTime(16, java.sql.Time(System.currentTimeMillis()))
                     st.setInt(17, 2023)
                     st.setString(18, "varcharValue$i")
@@ -268,8 +270,8 @@ class MariadbH2Test {
                     st.setDouble(11, i * 20.0)
                     st.setBigDecimal(12, BigDecimal(i * 20))
                     st.setDate(13, java.sql.Date(System.currentTimeMillis()))
-                    st.setTimestamp(14, java.sql.Timestamp(System.currentTimeMillis()))
-                    st.setTimestamp(15, java.sql.Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(14, Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(15, Timestamp(System.currentTimeMillis()))
                     st.setTime(16, java.sql.Time(System.currentTimeMillis()))
                     st.setInt(17, 2023)
                     st.setString(18, "varcharValue$i")
@@ -316,8 +318,8 @@ class MariadbH2Test {
         schema.columns["longblobcol"]!!.type shouldBe typeOf<Blob>()
         schema.columns["tinyblobcol"]!!.type shouldBe typeOf<Blob>()
         schema.columns["datecol"]!!.type shouldBe typeOf<Date>()
-        schema.columns["datetimecol"]!!.type shouldBe typeOf<java.sql.Timestamp>()
-        schema.columns["timestampcol"]!!.type shouldBe typeOf<java.sql.Timestamp>()
+        schema.columns["datetimecol"]!!.type shouldBe typeOf<Instant>()
+        schema.columns["timestampcol"]!!.type shouldBe typeOf<Instant>()
         schema.columns["timecol"]!!.type shouldBe typeOf<java.sql.Time>()
         schema.columns["yearcol"]!!.type shouldBe typeOf<Int>()
 
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mssqlH2Test.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mssqlH2Test.kt
index 3d37005b85..f625289fa4 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mssqlH2Test.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mssqlH2Test.kt
@@ -21,6 +21,7 @@ import java.sql.SQLException
 import java.util.Date
 import java.util.UUID
 import kotlin.reflect.typeOf
+import kotlin.time.Instant
 
 private const val URL =
     "jdbc:h2:mem:testmssql;DB_CLOSE_DELAY=-1;MODE=MSSQLServer;DATABASE_TO_UPPER=FALSE;CASE_INSENSITIVE_IDENTIFIERS=TRUE"
@@ -188,8 +189,8 @@ class MSSQLH2Test {
         schema.columns["bitColumn"]!!.type shouldBe typeOf<Boolean?>()
         schema.columns["charColumn"]!!.type shouldBe typeOf<String?>()
         schema.columns["dateColumn"]!!.type shouldBe typeOf<Date?>()
-        schema.columns["datetime3Column"]!!.type shouldBe typeOf<java.sql.Timestamp?>()
-        schema.columns["datetime2Column"]!!.type shouldBe typeOf<java.sql.Timestamp?>()
+        schema.columns["datetime3Column"]!!.type shouldBe typeOf<Instant?>()
+        schema.columns["datetime2Column"]!!.type shouldBe typeOf<Instant?>()
         schema.columns["decimalColumn"]!!.type shouldBe typeOf<BigDecimal?>()
         schema.columns["floatColumn"]!!.type shouldBe typeOf<Double?>()
         schema.columns["intColumn"]!!.type shouldBe typeOf<Int?>()
@@ -200,11 +201,11 @@ class MSSQLH2Test {
         schema.columns["nvarcharColumn"]!!.type shouldBe typeOf<String?>()
         schema.columns["nvarcharMaxColumn"]!!.type shouldBe typeOf<String?>()
         schema.columns["realColumn"]!!.type shouldBe typeOf<Float?>()
-        schema.columns["smalldatetimeColumn"]!!.type shouldBe typeOf<java.sql.Timestamp?>()
+        schema.columns["smalldatetimeColumn"]!!.type shouldBe typeOf<Instant?>()
         schema.columns["smallintColumn"]!!.type shouldBe typeOf<Int?>()
         schema.columns["smallmoneyColumn"]!!.type shouldBe typeOf<BigDecimal?>()
         schema.columns["timeColumn"]!!.type shouldBe typeOf<java.sql.Time?>()
-        schema.columns["timestampColumn"]!!.type shouldBe typeOf<java.sql.Timestamp?>()
+        schema.columns["timestampColumn"]!!.type shouldBe typeOf<Instant?>()
         schema.columns["tinyintColumn"]!!.type shouldBe typeOf<Int?>()
         schema.columns["varbinaryColumn"]!!.type shouldBe typeOf<ByteArray?>()
         schema.columns["varbinaryMaxColumn"]!!.type shouldBe typeOf<ByteArray?>()
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mysqlH2Test.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mysqlH2Test.kt
index c0a6cdbe14..86d6ebc5b2 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mysqlH2Test.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/mysqlH2Test.kt
@@ -20,8 +20,10 @@ import java.math.BigDecimal
 import java.sql.Connection
 import java.sql.DriverManager
 import java.sql.SQLException
+import java.sql.Timestamp
 import java.util.Date
 import kotlin.reflect.typeOf
+import kotlin.time.Instant
 
 // NOTE: the names of testing databases should be different to avoid collisions and should not contain the system names itself
 private const val URL = "jdbc:h2:mem:test2;DB_CLOSE_DELAY=-1;MODE=MySQL;DATABASE_TO_LOWER=TRUE"
@@ -228,8 +230,8 @@ class MySqlH2Test {
                     st.setDouble(11, i * 10.0)
                     st.setBigDecimal(12, BigDecimal(i * 10))
                     st.setDate(13, java.sql.Date(System.currentTimeMillis()))
-                    st.setTimestamp(14, java.sql.Timestamp(System.currentTimeMillis()))
-                    st.setTimestamp(15, java.sql.Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(14, Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(15, Timestamp(System.currentTimeMillis()))
                     st.setTime(16, java.sql.Time(System.currentTimeMillis()))
                     st.setInt(17, 2023)
                     st.setString(18, "varcharValue$i")
@@ -265,8 +267,8 @@ class MySqlH2Test {
                     st.setDouble(11, i * 20.0)
                     st.setBigDecimal(12, BigDecimal(i * 20))
                     st.setDate(13, java.sql.Date(System.currentTimeMillis()))
-                    st.setTimestamp(14, java.sql.Timestamp(System.currentTimeMillis()))
-                    st.setTimestamp(15, java.sql.Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(14, Timestamp(System.currentTimeMillis()))
+                    st.setTimestamp(15, Timestamp(System.currentTimeMillis()))
                     st.setTime(16, java.sql.Time(System.currentTimeMillis()))
                     st.setInt(17, 2023)
                     st.setString(18, "varcharValue$i")
@@ -308,8 +310,8 @@ class MySqlH2Test {
         schema.columns["id"]!!.type shouldBe typeOf<Int>()
         schema.columns["textcol"]!!.type shouldBe typeOf<String>()
         schema.columns["datecol"]!!.type shouldBe typeOf<Date>()
-        schema.columns["datetimecol"]!!.type shouldBe typeOf<java.sql.Timestamp>()
-        schema.columns["timestampcol"]!!.type shouldBe typeOf<java.sql.Timestamp>()
+        schema.columns["datetimecol"]!!.type shouldBe typeOf<Instant>()
+        schema.columns["timestampcol"]!!.type shouldBe typeOf<Instant>()
         schema.columns["timecol"]!!.type shouldBe typeOf<java.sql.Time>()
         schema.columns["yearcol"]!!.type shouldBe typeOf<Int>()
         schema.columns["varbinarycol"]!!.type shouldBe typeOf<ByteArray>()
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/postgresH2Test.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/postgresH2Test.kt
index d2b4bef65d..efd1425555 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/postgresH2Test.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/h2/postgresH2Test.kt
@@ -123,6 +123,18 @@ class PostgresH2Test {
 
             connection.createStatement().execute(createTableQuery.trimIndent())
 
+            // Table with STRUCT/ROW column to verify JDBC Types.STRUCT handling
+            @Language("SQL")
+            val createTableWithStruct =
+                """
+                CREATE TABLE IF NOT EXISTS table3 (
+                    id serial PRIMARY KEY,
+                    structCol ROW(a INT, b VARCHAR(10)) not null
+                )
+                """.trimIndent()
+
+            connection.createStatement().execute(createTableWithStruct)
+
             @Language("SQL")
             val insertData1 =
                 """
@@ -196,6 +208,17 @@ class PostgresH2Test {
                     st.executeUpdate()
                 }
             }
+
+            // Insert data into table3 with ROW/STRUCT literals
+            @Language("SQL")
+            val insertStructs =
+                """
+                INSERT INTO table3 (structCol) VALUES
+                (ROW(1, 'X')),
+                (ROW(2, 'Y')),
+                (ROW(3, 'Z'))
+                """.trimIndent()
+            connection.createStatement().execute(insertStructs)
         }
 
         @AfterClass
@@ -286,6 +309,19 @@ class PostgresH2Test {
         table2Df[0][4] shouldBe 1001
     }
 
+    @Test
+    fun `read composite column from table`() {
+        val tableName3 = "table3"
+        val df3 = DataFrame.readSqlTable(connection, tableName3)
+
+        // Validate row count
+        df3.rowsCount() shouldBe 3
+
+        // Validate schema type stays as Any for STRUCT (no special mapping yet)
+        val schema3 = DataFrameSchema.readSqlTable(connection, tableName3)
+        schema3.columns["structcol"]!!.type shouldBe typeOf<Any>()
+    }
+
     @Test
     fun `read columns of different types to check type mapping`() {
         val tableName1 = "table1"
diff --git a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/local/duckDbTest.kt b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/local/duckDbTest.kt
index 3cc00e15a3..e0e324654a 100644
--- a/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/local/duckDbTest.kt
+++ b/dataframe-jdbc/src/test/kotlin/org/jetbrains/kotlinx/dataframe/io/local/duckDbTest.kt
@@ -2,7 +2,10 @@
 
 package org.jetbrains.kotlinx.dataframe.io.local
 
+import io.kotest.assertions.withClue
 import io.kotest.matchers.shouldBe
+import kotlinx.datetime.LocalDate
+import kotlinx.datetime.LocalTime
 import org.duckdb.DuckDBConnection
 import org.duckdb.DuckDBResultSet
 import org.duckdb.JsonNode
@@ -10,9 +13,11 @@ import org.jetbrains.kotlinx.dataframe.AnyFrame
 import org.jetbrains.kotlinx.dataframe.DataFrame
 import org.jetbrains.kotlinx.dataframe.annotations.ColumnName
 import org.jetbrains.kotlinx.dataframe.annotations.DataSchema
+import org.jetbrains.kotlinx.dataframe.api.DataRowSchema
 import org.jetbrains.kotlinx.dataframe.api.cast
 import org.jetbrains.kotlinx.dataframe.api.colsOf
 import org.jetbrains.kotlinx.dataframe.api.convert
+import org.jetbrains.kotlinx.dataframe.api.dataFrameOf
 import org.jetbrains.kotlinx.dataframe.api.reorderColumnsByName
 import org.jetbrains.kotlinx.dataframe.api.schema
 import org.jetbrains.kotlinx.dataframe.api.single
@@ -36,11 +41,12 @@ import java.nio.file.Files
 import java.sql.Blob
 import java.sql.DriverManager
 import java.sql.Timestamp
-import java.time.LocalDate
-import java.time.LocalTime
-import java.time.OffsetDateTime
 import java.util.UUID
 import kotlin.io.path.createTempDirectory
+import kotlin.time.Instant
+import kotlin.time.toKotlinInstant
+import kotlin.uuid.Uuid
+import java.time.OffsetDateTime as JavaOffsetDateTime
 
 private const val URL = "jdbc:duckdb:"
 
@@ -58,10 +64,10 @@ class DuckDbTest {
     ) {
         companion object {
             val expected = listOf(
-                Person(1, "John Doe", 30, 50000.0, LocalDate.of(2020, 1, 15)),
-                Person(2, "Jane Smith", 28, 55000.0, LocalDate.of(2021, 3, 20)),
-                Person(3, "Bob Johnson", 35, 65000.0, LocalDate.of(2019, 11, 10)),
-                Person(4, "Alice Brown", 32, 60000.0, LocalDate.of(2020, 7, 1)),
+                Person(1, "John Doe", 30, 50000.0, LocalDate(2020, 1, 15)),
+                Person(2, "Jane Smith", 28, 55000.0, LocalDate(2021, 3, 20)),
+                Person(3, "Bob Johnson", 35, 65000.0, LocalDate(2019, 11, 10)),
+                Person(4, "Alice Brown", 32, 60000.0, LocalDate(2020, 7, 1)),
             ).toDataFrame()
         }
     }
@@ -91,7 +97,7 @@ class DuckDbTest {
         @ColumnName("date_col")
         val dateCol: LocalDate,
         @ColumnName("datetime_col")
-        val datetimeCol: Timestamp,
+        val datetimeCol: Instant,
         @ColumnName("decimal_col")
         val decimalCol: BigDecimal,
         @ColumnName("double_col")
@@ -151,11 +157,11 @@ class DuckDbTest {
         @ColumnName("time_col")
         val timeCol: LocalTime,
         @ColumnName("timestamp_col")
-        val timestampCol: Timestamp,
+        val timestampCol: Instant,
         @ColumnName("timestamptz_col")
-        val timestamptzCol: OffsetDateTime,
+        val timestamptzCol: JavaOffsetDateTime,
         @ColumnName("timestampwtz_col")
-        val timestampwtzCol: OffsetDateTime,
+        val timestampwtzCol: JavaOffsetDateTime,
         @ColumnName("tinyint_col")
         val tinyintCol: Byte,
         @ColumnName("ubigint_col")
@@ -179,7 +185,7 @@ class DuckDbTest {
         @ColumnName("utinyint_col")
         val utinyintCol: Short,
         @ColumnName("uuid_col")
-        val uuidCol: UUID,
+        val uuidCol: Uuid,
         @ColumnName("varbinary_col")
         val varbinaryCol: Blob,
         @ColumnName("varchar_col")
@@ -199,7 +205,7 @@ class DuckDbTest {
                     byteaCol = DuckDBResultSet.DuckDBBlobResult(ByteBuffer.wrap("DEADBEEF".toByteArray())),
                     charCol = "test",
                     dateCol = LocalDate.parse("2025-06-19"),
-                    datetimeCol = Timestamp.valueOf("2025-06-19 12:34:56"),
+                    datetimeCol = Timestamp.valueOf("2025-06-19 12:34:56").toInstant().toKotlinInstant(),
                     decimalCol = BigDecimal("123.45"),
                     doubleCol = 3.14159,
                     enumCol = "female",
@@ -229,9 +235,9 @@ class DuckDbTest {
                     stringCol = "test string",
                     textCol = "test text",
                     timeCol = LocalTime.parse("12:34:56"),
-                    timestampCol = Timestamp.valueOf("2025-06-19 12:34:56"),
-                    timestamptzCol = OffsetDateTime.parse("2025-06-19T12:34:56+02:00"),
-                    timestampwtzCol = OffsetDateTime.parse("2025-06-19T12:34:56+02:00"),
+                    timestampCol = Timestamp.valueOf("2025-06-19 12:34:56").toInstant().toKotlinInstant(),
+                    timestamptzCol = JavaOffsetDateTime.parse("2025-06-19T12:34:56+02:00"),
+                    timestampwtzCol = JavaOffsetDateTime.parse("2025-06-19T12:34:56+02:00"),
                     tinyintCol = 127,
                     ubigintCol = BigInteger("18446744073709551615"),
                     uhugeintCol = BigInteger("340282366920938463463374607431768211455"),
@@ -243,7 +249,7 @@ class DuckDbTest {
                     uintCol = 4294967295L,
                     usmallintCol = 65535,
                     utinyintCol = 255,
-                    uuidCol = UUID.fromString("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"),
+                    uuidCol = Uuid.parse("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"),
                     varbinaryCol = DuckDBResultSet.DuckDBBlobResult(ByteBuffer.wrap("DEADBEEF".toByteArray())),
                     varcharCol = "test string",
                 ),
@@ -251,24 +257,34 @@ class DuckDbTest {
         }
     }
 
+    @DataSchema
+    data class NestedEntry(val i: Int, val j: String) : DataRowSchema
+
+    @DataSchema
+    data class NullableNestedEntry(val i: Int?, val j: String?) : DataRowSchema
+
     @DataSchema
     data class NestedTypes(
+        @ColumnName("test_col")
+        val testCol: Int,
         @ColumnName("ijstruct_col")
-        val ijstructCol: java.sql.Struct,
+        val ijstructCol: NestedEntry,
+        @ColumnName("ijstructlist_col")
+        val ijstructlistCol: DataFrame<NullableNestedEntry>,
         @ColumnName("intarray_col")
-        val intarrayCol: java.sql.Array,
+        val intarrayCol: List<Int?>,
         @ColumnName("intlist_col")
-        val intlistCol: java.sql.Array,
+        val intlistCol: List<Int?>,
         @ColumnName("intstringmap_col")
         val intstringmapCol: Map<Int, String?>,
         @ColumnName("intstrinstinggmap_col")
         val intstrinstinggmapCol: Map<Int, Map<String, String?>?>,
         @ColumnName("stringarray_col")
-        val stringarrayCol: java.sql.Array,
+        val stringarrayCol: List<String?>,
         @ColumnName("stringlist_col")
-        val stringlistCol: java.sql.Array,
+        val stringlistCol: List<String?>,
         @ColumnName("stringlistlist_col")
-        val stringlistlistCol: java.sql.Array,
+        val stringlistlistCol: List<List<String?>?>,
         @ColumnName("union_col")
         val unionCol: Any,
     )
@@ -310,7 +326,19 @@ class DuckDbTest {
             subset = DataFrame.readSqlQuery(connection, """SELECT test_table.name, test_table.age FROM test_table""")
         }
 
-        schema.compare(Person.expected.schema()).isSuperOrMatches() shouldBe true
+        withClue({
+            """
+            |Read schema must be <: expected schema
+            |
+            |Read Schema: 
+            |${schema.toString().lines().joinToString("\n|")}
+            |
+            |expected Schema:
+            |${Person.expected.schema().toString().lines().joinToString("\n|")}
+            """.trimMargin()
+        }) {
+            schema.compare(Person.expected.schema()).isSuperOrMatches() shouldBe true
+        }
 
         df.cast<Person>(verify = true) shouldBe Person.expected
         df.assertInferredTypesMatchSchema()
@@ -545,10 +573,24 @@ class DuckDbTest {
             df = DataFrame.readSqlTable(connection, "table1").reorderColumnsByName()
         }
 
-        schema.compare(GeneralPurposeTypes.expected.schema()).isSuperOrMatches() shouldBe true
+//        schema.toString().lines().sorted().joinToString("\n") shouldBe
+//            GeneralPurposeTypes.expected.schema().toString().lines().sorted().joinToString("\n")
+        withClue({
+            """
+            |Read schema must be <: expected schema
+            |
+            |Read Schema: 
+            |${schema.toString().lines().joinToString("\n|")}
+            |
+            |expected Schema:
+            |${GeneralPurposeTypes.expected.schema().toString().lines().joinToString("\n|")}
+            """.trimMargin()
+        }) {
+            schema.compare(GeneralPurposeTypes.expected.schema()).isSuperOrMatches() shouldBe true
+        }
 
         // on some systems OffsetDateTime's get converted to UTC sometimes, let's compare them as Instant instead
-        fun AnyFrame.fixOffsetDateTime() = convert { colsOf<OffsetDateTime>() }.with { it.toInstant() }
+        fun AnyFrame.fixOffsetDateTime() = convert { colsOf<JavaOffsetDateTime>() }.with { it.toInstant() }
 
         df.cast<GeneralPurposeTypes>(verify = true).fixOffsetDateTime() shouldBe
             GeneralPurposeTypes.expected.fixOffsetDateTime()
@@ -567,6 +609,7 @@ class DuckDbTest {
             connection.prepareStatement(
                 """
                 CREATE TABLE IF NOT EXISTS table2 (
+                    test_col INTEGER,
                     intarray_col INTEGER[3],
                     stringarray_col VARCHAR[3],
                     intlist_col INTEGER[],
@@ -575,6 +618,7 @@ class DuckDbTest {
                     intstringmap_col MAP(INTEGER, VARCHAR),
                     intstrinstinggmap_col MAP(INTEGER, MAP(VARCHAR, VARCHAR)),
                     ijstruct_col STRUCT(i INTEGER, j VARCHAR),
+                    ijstructlist_col STRUCT(i INTEGER, j VARCHAR)[],
                     union_col UNION(num INTEGER, text VARCHAR),
                 )
                 """.trimIndent(),
@@ -583,6 +627,7 @@ class DuckDbTest {
             connection.prepareStatement(
                 """
                 INSERT INTO table2 VALUES (
+                    1,                                      -- int
                     array_value(1, 2, NULL),                    -- int array
                     array_value('a', 'ab', 'abc'),           -- string array
                     list_value(1, 2, 3),                     -- int list
@@ -591,6 +636,7 @@ class DuckDbTest {
                     MAP { 1: 'value1', 200: 'value2' },      -- int string map
                     MAP { 1: MAP { 'value1': 'a', 'value2': 'b' }, 200: MAP { 'value1': 'c', 'value2': 'd' } }, -- int string string map
                     { 'i': 42, 'j': 'answer' },               -- struct
+                    list_value({ 'i': 42, 'j': 'answer' }, { 'i': 44, 'j': 'answer' }), -- struct list
                     union_value(num := 2),                    -- union
                 )
                 """.trimIndent(),
@@ -606,19 +652,24 @@ class DuckDbTest {
         df as DataFrame<NestedTypes>
 
         df.single().let {
-            it[{ "intarray_col"<java.sql.Array>() }].array shouldBe arrayOf(1, 2, null)
-            it[{ "stringarray_col"<java.sql.Array>() }].array shouldBe arrayOf("a", "ab", "abc")
-            it[{ "intlist_col"<java.sql.Array>() }].array shouldBe arrayOf(1, 2, 3)
-            it[{ "stringlist_col"<java.sql.Array>() }].array shouldBe arrayOf("a", "ab", "abc")
-            (it[{ "stringlistlist_col"<java.sql.Array>() }].array as Array<*>)
-                .map { (it as java.sql.Array?)?.array } shouldBe listOf(arrayOf("a", "ab"), arrayOf("abc"), null)
-            it[{ "intstringmap_col"<Map<Int, String?>>() }] shouldBe mapOf(1 to "value1", 200 to "value2")
-            it[{ "intstrinstinggmap_col"<Map<Int, Map<String, String?>>>() }] shouldBe mapOf(
+            it["intarray_col"] shouldBe listOf(1, 2, null)
+            it["stringarray_col"] shouldBe listOf("a", "ab", "abc")
+            it["intlist_col"] shouldBe listOf(1, 2, 3)
+            it["stringlist_col"] shouldBe listOf("a", "ab", "abc")
+            it["stringlistlist_col"] shouldBe listOf(listOf("a", "ab"), listOf("abc"), null)
+            it["intstringmap_col"] shouldBe mapOf(1 to "value1", 200 to "value2")
+            it["intstrinstinggmap_col"] shouldBe mapOf(
                 1 to mapOf("value1" to "a", "value2" to "b"),
                 200 to mapOf("value1" to "c", "value2" to "d"),
             )
-            it[{ "ijstruct_col"<java.sql.Struct>() }].attributes shouldBe arrayOf<Any>(42, "answer")
-            it[{ "union_col"<Any>() }] shouldBe 2
+            it[{ "ijstruct_col"["i"]<Int>() }] shouldBe 42
+            it[{ "ijstruct_col"["j"]<String>() }] shouldBe "answer"
+            it[{ "ijstructlist_col"<DataFrame<NestedEntry>>() }] shouldBe
+                dataFrameOf(
+                    NestedEntry(42, "answer"),
+                    NestedEntry(44, "answer"),
+                )
+            it["union_col"] shouldBe 2
         }
     }
 
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 6b0513fb39..91dd09576d 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -54,7 +54,7 @@ android-gradle-api = "7.3.1" # need to revise our tests to update
 ktor = "3.0.1" # needs jupyter compatibility with Kotlin 2.1 to update
 kotlin-compile-testing = "0.7.1"
 hikari = "7.0.2"
-duckdb = "1.3.1.0"
+duckdb = "1.4.2.0"
 buildconfig = "5.6.7"
 benchmark = "0.4.12"