diff --git a/buildSrc/private/src/main/kotlin/androidx/build/uptodatedness/TaskUpToDateValidator.kt b/buildSrc/private/src/main/kotlin/androidx/build/uptodatedness/TaskUpToDateValidator.kt
index b04ea44..074b07d 100644
--- a/buildSrc/private/src/main/kotlin/androidx/build/uptodatedness/TaskUpToDateValidator.kt
+++ b/buildSrc/private/src/main/kotlin/androidx/build/uptodatedness/TaskUpToDateValidator.kt
@@ -185,6 +185,7 @@
     "lintVitalDebug",
     "lintWithExpandProjectionDebug",
     "lintWithoutExpandProjectionDebug",
+    "lintWithNullAwareTypeConverterDebug",
     "lintWithKaptDebug",
     "lintWithKspDebug",
 
diff --git a/room/integration-tests/kotlintestapp/src/androidTestWithKsp/java/androidx/room/integration/kotlintestapp/NullabilityAwareTypeConversionTest.kt b/room/integration-tests/kotlintestapp/src/androidTestWithKsp/java/androidx/room/integration/kotlintestapp/NullabilityAwareTypeConversionTest.kt
new file mode 100644
index 0000000..85c86ea
--- /dev/null
+++ b/room/integration-tests/kotlintestapp/src/androidTestWithKsp/java/androidx/room/integration/kotlintestapp/NullabilityAwareTypeConversionTest.kt
@@ -0,0 +1,357 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.integration.kotlintestapp
+
+import android.database.Cursor
+import androidx.room.Dao
+import androidx.room.Database
+import androidx.room.Entity
+import androidx.room.Insert
+import androidx.room.PrimaryKey
+import androidx.room.ProvidedTypeConverter
+import androidx.room.Query
+import androidx.room.Room
+import androidx.room.RoomDatabase
+import androidx.room.TypeConverter
+import androidx.room.TypeConverters
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.filters.SmallTest
+import com.google.common.truth.Truth.assertThat
+import com.google.common.truth.Truth.assertWithMessage
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+
+/**
+ * This test can only pass in KSP with the new type converter store, which is why it is only in the
+ * KSP specific source set.
+ */
+@RunWith(AndroidJUnit4::class)
+@SmallTest
+class NullabilityAwareTypeConversionTest {
+    lateinit var dao: UserDao
+    private val nullableConvertors = NullableTypeConverters()
+
+    @Before
+    fun init() {
+        dao = Room.inMemoryDatabaseBuilder(
+            ApplicationProvider.getApplicationContext(),
+            NullAwareConverterDatabase::class.java
+        ).addTypeConverter(nullableConvertors).build().userDao
+    }
+
+    private fun assertNullableConverterIsNotUsed() {
+        assertWithMessage(
+            "should've not used nullable conversion since it is not available in this scope"
+        ).that(nullableConvertors.toStringInvocations).isEmpty()
+        assertWithMessage(
+            "should've not used nullable conversion since it is not available in this scope"
+        ).that(nullableConvertors.fromStringInvocations).isEmpty()
+    }
+
+    @Test
+    fun insert() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = Country.UNITED_KINGDOM
+        )
+        dao.insert(user)
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-FR-UK")
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun setNonNullColumn() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-FR-null")
+        dao.setNonNullCountry(id = 1, nonNullCountry = Country.UNITED_KINGDOM)
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-UK-null")
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun setNullableColumn() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = Country.UNITED_KINGDOM
+        )
+        dao.insert(user)
+        dao.setNullableCountry(id = 1, nullableCountry = null)
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-FR-null")
+        dao.setNullableCountry(id = 1, nullableCountry = Country.UNITED_KINGDOM)
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-FR-UK")
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun load() {
+        val user1 = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = Country.UNITED_KINGDOM
+        )
+        dao.insert(user1)
+        assertThat(
+            dao.getById(1)
+        ).isEqualTo(user1)
+        val user2 = User(
+            id = 2,
+            nonNullCountry = Country.UNITED_KINGDOM,
+            nullableCountry = null
+        )
+        dao.insert(user2)
+        assertThat(
+            dao.getById(2)
+        ).isEqualTo(user2)
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun useNullableConverter() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = Country.UNITED_KINGDOM
+        )
+        dao.insert(user)
+        dao.setNullableCountryWithNullableTypeConverter(
+            id = 1,
+            nullableCountry = null
+        )
+        assertThat(
+            dao.getRawData()
+        ).isEqualTo("1-FR-null")
+        assertThat(
+            nullableConvertors.toStringInvocations
+        ).containsExactly(null)
+    }
+
+    @Test
+    fun loadNonNullColumn() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        val country = dao.getNonNullCountry(id = 1)
+        assertThat(country).isEqualTo(Country.FRANCE)
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun loadNullableColumn() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        val country = dao.getNullableCountry(id = 1)
+        assertThat(country).isNull()
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun loadNonNullColumn_withNullableConverter() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        val country = dao.getNonNullCountryWithNullableTypeConverter(id = 1)
+        assertThat(country).isEqualTo(Country.FRANCE)
+        // return value is non-null so it is better to use non-null converter and assume
+        // column is non-null, instead of using the nullable converter
+        assertNullableConverterIsNotUsed()
+    }
+
+    @Test
+    fun loadNonNullColumn_asNullable_withNullableConverter() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        val country = dao.getNonNullCountryAsNullableWithNullableTypeConverter(id = 1)
+        assertThat(country).isEqualTo(Country.FRANCE)
+        // return value is nullable so we are using the nullable converter because room does not
+        // know that the column is non-null.
+        // if one day Room understands it and this test fails, feel free to update it.
+        // We still want this test because right now Room does not know column is non-null hence
+        // it should prefer the nullable converter.
+        assertThat(
+            nullableConvertors.fromStringInvocations
+        ).containsExactly("FR")
+    }
+
+    @Test
+    fun loadNullableColumn_withNullableConverter() {
+        val user = User(
+            id = 1,
+            nonNullCountry = Country.FRANCE,
+            nullableCountry = null
+        )
+        dao.insert(user)
+        val country = dao.getNullableCountryWithNullableTypeConverter(id = 1)
+        assertThat(country).isNull()
+        assertThat(
+            nullableConvertors.fromStringInvocations
+        ).containsExactly(null)
+    }
+
+    @Database(
+        version = 1,
+        entities = [
+            User::class,
+        ],
+        exportSchema = false
+    )
+    @TypeConverters(NonNullTypeConverters::class)
+    abstract class NullAwareConverterDatabase : RoomDatabase() {
+        abstract val userDao: UserDao
+    }
+
+    @Dao
+    abstract class UserDao {
+
+        @Insert
+        abstract fun insert(user: User): Long
+
+        @Query("UPDATE user SET nonNullCountry = :nonNullCountry WHERE id = :id")
+        abstract fun setNonNullCountry(id: Long, nonNullCountry: Country)
+
+        @Query("UPDATE user SET nullableCountry = :nullableCountry WHERE id = :id")
+        abstract fun setNullableCountry(id: Long, nullableCountry: Country?)
+
+        @Query("SELECT * FROM user WHERE id = :id")
+        abstract fun getById(id: Long): User?
+
+        @Query("UPDATE user SET nullableCountry = :nullableCountry WHERE id = :id")
+        @TypeConverters(NullableTypeConverters::class)
+        abstract fun setNullableCountryWithNullableTypeConverter(
+            id: Long,
+            nullableCountry: Country?
+        )
+
+        @Query("SELECT nonNullCountry FROM user WHERE id = :id")
+        abstract fun getNonNullCountry(id: Long): Country
+
+        @Query("SELECT nullableCountry FROM user WHERE id = :id")
+        abstract fun getNullableCountry(id: Long): Country?
+
+        @Query("SELECT nullableCountry FROM user WHERE id = :id")
+        @TypeConverters(NullableTypeConverters::class)
+        abstract fun getNullableCountryWithNullableTypeConverter(id: Long): Country?
+
+        @Query("SELECT nonNullCountry FROM user WHERE id = :id")
+        @TypeConverters(NullableTypeConverters::class)
+        abstract fun getNonNullCountryWithNullableTypeConverter(id: Long): Country
+
+        @Query("SELECT nonNullCountry FROM user WHERE id = :id")
+        @TypeConverters(NullableTypeConverters::class)
+        abstract fun getNonNullCountryAsNullableWithNullableTypeConverter(id: Long): Country?
+
+        @Query("SELECT * FROM User ORDER BY id")
+        protected abstract fun getUsers(): Cursor
+
+        /**
+         * Return raw data in the database so that we can assert what is in the database
+         * without room's converters
+         */
+        fun getRawData(): String {
+            return buildString {
+                getUsers().use {
+                    if (it.moveToNext()) {
+                        append(it.getInt(0))
+                        append("-")
+                        append(it.getString(1))
+                        append("-")
+                        append(it.getString(2))
+                    }
+                }
+            }
+        }
+    }
+
+    @Entity(tableName = "user")
+    data class User(
+        @PrimaryKey
+        val id: Long,
+        val nonNullCountry: Country,
+        val nullableCountry: Country?,
+    )
+
+    enum class Country(val countryCode: String) {
+        UNITED_KINGDOM("UK"),
+        FRANCE("FR"),
+    }
+
+    object NonNullTypeConverters {
+        @TypeConverter
+        fun toString(country: Country): String {
+            return country.countryCode
+        }
+
+        @TypeConverter
+        fun toCountry(string: String): Country {
+            return Country.values().find { it.countryCode == string }
+                ?: throw IllegalArgumentException("Country code '$string' not found")
+        }
+    }
+
+    @ProvidedTypeConverter
+    class NullableTypeConverters {
+        val toStringInvocations = mutableListOf<Country?>()
+        val fromStringInvocations = mutableListOf<String?>()
+        @TypeConverter
+        fun toString(country: Country?): String? {
+            toStringInvocations.add(country)
+            return country?.countryCode
+        }
+
+        @TypeConverter
+        fun toCountry(string: String?): Country? {
+            fromStringInvocations.add(string)
+            if (string == null) {
+                return null
+            }
+            return Country.values().find { it.countryCode == string }
+                ?: throw IllegalArgumentException("Country code '$string' not found")
+        }
+    }
+}
diff --git a/room/integration-tests/testapp/build.gradle b/room/integration-tests/testapp/build.gradle
index 17cf821..358a6e5 100644
--- a/room/integration-tests/testapp/build.gradle
+++ b/room/integration-tests/testapp/build.gradle
@@ -68,6 +68,18 @@
                 }
             }
         }
+        withNullAwareTypeConverter {
+            dimension "processorConfiguration"
+            javaCompileOptions {
+                annotationProcessorOptions {
+                    arguments = [
+                            "room.schemaLocation"  : "$projectDir/schemas".toString(),
+                            "room.expandProjection"  : "false",
+                            "room.useNullAwareTypeAnalysis": "true"
+                    ]
+                }
+            }
+        }
     }
 }
 
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/RoomKspProcessor.kt b/room/room-compiler/src/main/kotlin/androidx/room/RoomKspProcessor.kt
index 7437810..0733517 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/RoomKspProcessor.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/RoomKspProcessor.kt
@@ -17,9 +17,12 @@
 package androidx.room
 
 import androidx.room.compiler.processing.ksp.KspBasicAnnotationProcessor
+import androidx.room.processor.Context.BooleanProcessorOptions.USE_NULL_AWARE_CONVERTER
+import androidx.room.processor.ProcessorErrors
 import com.google.devtools.ksp.processing.SymbolProcessor
 import com.google.devtools.ksp.processing.SymbolProcessorEnvironment
 import com.google.devtools.ksp.processing.SymbolProcessorProvider
+import javax.tools.Diagnostic
 
 /**
  * Entry point for processing using KSP.
@@ -27,7 +30,22 @@
 class RoomKspProcessor(
     environment: SymbolProcessorEnvironment
 ) : KspBasicAnnotationProcessor(environment) {
-
+    init {
+        // print a warning if null aware converter is disabled because we'll remove that ability
+        // soon.
+        if (USE_NULL_AWARE_CONVERTER.getInputValue(xProcessingEnv) == false) {
+            xProcessingEnv.messager.printMessage(
+                kind = Diagnostic.Kind.WARNING,
+                msg = """
+                    Disabling null-aware type analysis in KSP is a temporary flag that will be
+                    removed in a future release.
+                    If the null-aware type analysis is causing a bug in your application,
+                    please file a bug at ${ProcessorErrors.ISSUE_TRACKER_LINK} with
+                    a sample app that reproduces your problem.
+                """.trimIndent()
+            )
+        }
+    }
     override fun processingSteps() = listOf(
         DatabaseProcessingStep()
     )
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/processor/Context.kt b/room/room-compiler/src/main/kotlin/androidx/room/processor/Context.kt
index 537ceff..791a90e 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/processor/Context.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/processor/Context.kt
@@ -41,6 +41,15 @@
     val checker: Checks = Checks(logger)
     val COMMON_TYPES = CommonTypes(processingEnv)
 
+    /**
+     * Checks whether we should use the TypeConverter store that has a specific heuristic for
+     * nullability. Defaults to true in KSP, false in javac.
+     */
+    val useNullAwareConverter: Boolean by lazy {
+        BooleanProcessorOptions.USE_NULL_AWARE_CONVERTER.getInputValue(processingEnv)
+            ?: (processingEnv.backend == XProcessingEnv.Backend.KSP)
+    }
+
     val typeAdapterStore by lazy {
         if (inheritedAdapterStore != null) {
             TypeAdapterStore.copy(this, inheritedAdapterStore)
@@ -207,19 +216,21 @@
 
     enum class BooleanProcessorOptions(val argName: String, private val defaultValue: Boolean) {
         INCREMENTAL("room.incremental", defaultValue = true),
-        EXPAND_PROJECTION("room.expandProjection", defaultValue = false);
+        EXPAND_PROJECTION("room.expandProjection", defaultValue = false),
+        USE_NULL_AWARE_CONVERTER("room.useNullAwareTypeAnalysis", defaultValue = false);
 
         /**
          * Returns the value of this option passed through the [XProcessingEnv]. If the value
          * is null or blank, it returns the default value instead.
          */
         fun getValue(processingEnv: XProcessingEnv): Boolean {
-            val value = processingEnv.options[argName]
-            return if (value.isNullOrBlank()) {
-                defaultValue
-            } else {
-                value.toBoolean()
-            }
+            return getInputValue(processingEnv) ?: defaultValue
+        }
+
+        fun getInputValue(processingEnv: XProcessingEnv): Boolean? {
+            return processingEnv.options[argName]?.takeIf {
+                it.isNotBlank()
+            }?.toBoolean()
         }
     }
 }
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/NullAwareTypeConverterStore.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/NullAwareTypeConverterStore.kt
new file mode 100644
index 0000000..95e4a8f
--- /dev/null
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/NullAwareTypeConverterStore.kt
@@ -0,0 +1,429 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver
+
+import androidx.room.compiler.processing.XNullability
+import androidx.room.compiler.processing.XNullability.NONNULL
+import androidx.room.compiler.processing.XNullability.NULLABLE
+import androidx.room.compiler.processing.XNullability.UNKNOWN
+import androidx.room.compiler.processing.XProcessingEnv.Backend
+import androidx.room.compiler.processing.XType
+import androidx.room.processor.Context
+import androidx.room.solver.types.CompositeTypeConverter
+import androidx.room.solver.types.NoOpConverter
+import androidx.room.solver.types.NullSafeTypeConverter
+import androidx.room.solver.types.RequireNotNullTypeConverter
+import androidx.room.solver.types.TypeConverter
+import androidx.room.solver.types.UpCastTypeConverter
+import java.util.PriorityQueue
+
+/**
+ * A [TypeConverterStore] implementation that generates better code when we have the nullability
+ * information in types. It is enabled by default only in KSP backend but it can also be turned
+ * on via the [Context.BooleanProcessorOptions.USE_NULL_AWARE_CONVERTER] flag.
+ *
+ * This [TypeConverterStore] tries to maintain the nullability of the input/output type
+ * when writing into/reading from database. Even though nullability preservation is preferred, it is
+ * not strictly required such that it will fall back to the mismatched nullability.
+ */
+class NullAwareTypeConverterStore(
+    context: Context,
+    /**
+     * Available TypeConverters. Note that we might synthesize new type converters based on this
+     * list.
+     */
+    typeConverters: List<TypeConverter>,
+    /**
+     * List of types that can be saved into db/read from without a converter.
+     */
+    private val knownColumnTypes: List<XType>
+) : TypeConverterStore {
+    override val typeConverters = if (context.processingEnv.backend == Backend.KSP) {
+        val processedConverters = typeConverters.toMutableList()
+        // create copies for converters that receive non-null values
+        typeConverters.forEach { converter ->
+            if (converter.from.nullability == NONNULL) {
+                val candidate = NullSafeTypeConverter(delegate = converter)
+                // before we add this null safe converter, make sure there is no other converter
+                // that would already handle the same arguments.
+                val match = processedConverters.any { other ->
+                    other.from.isAssignableFrom(candidate.from) &&
+                        candidate.to.isAssignableFrom(other.to)
+                }
+                if (!match) {
+                    processedConverters.add(candidate)
+                }
+            }
+        }
+        processedConverters
+    } else {
+        typeConverters
+    }
+
+    // cache for type converter lookups to avoid traversing all of the list every time we need to
+    // find possible converters for a type. Unlike JAVAC, KSP supports equality in its objects so
+    // this tends to work rather well.
+    private val typeConvertersByFromCache = mutableMapOf<XType, List<TypeConverter>>()
+    private val typeConvertersByToCache = mutableMapOf<XType, List<TypeConverter>>()
+
+    /**
+     * Known column types that are nullable.
+     * Used in [getColumnTypesInPreferenceBuckets] to avoid re-partitioning known type lists.
+     */
+    private val knownNullableColumnTypes by lazy {
+        knownColumnTypes.filter { it.nullability == NULLABLE }
+    }
+
+    /**
+     * Known column types that are non-null or have unknown nullability.
+     * Used in [getColumnTypesInPreferenceBuckets] to avoid re-partitioning known type lists.
+     */
+    private val knownNonNullableColumnTypes by lazy {
+        knownColumnTypes.filter { it.nullability != NULLABLE }
+    }
+
+    /**
+     * Returns a list of lists for the given type, ordered by preference buckets for
+     * the given nullability.
+     */
+    private fun getColumnTypesInPreferenceBuckets(
+        nullability: XNullability,
+        explicitColumnTypes: List<XType>?
+    ): List<List<XType>> {
+        return if (explicitColumnTypes == null) {
+            when (nullability) {
+                NULLABLE -> {
+                    // prioritize nulls
+                    listOf(
+                        knownNullableColumnTypes,
+                        knownNonNullableColumnTypes
+                    )
+                }
+                NONNULL -> {
+                    // prioritize non-null
+                    listOf(
+                        knownNonNullableColumnTypes,
+                        knownNullableColumnTypes
+                    )
+                }
+                else -> {
+                    // we don't know, YOLO
+                    listOf(knownColumnTypes)
+                }
+            }
+        } else {
+            when (nullability) {
+                UNKNOWN -> listOf(explicitColumnTypes)
+                else -> listOf(
+                    explicitColumnTypes.filter { it.nullability == nullability },
+                    explicitColumnTypes.filter { it.nullability != nullability }
+                )
+            }
+        }
+    }
+
+    override fun findConverterIntoStatement(
+        input: XType,
+        columnTypes: List<XType>?
+    ): TypeConverter? {
+        getColumnTypesInPreferenceBuckets(
+            nullability = input.nullability,
+            explicitColumnTypes = columnTypes
+        ).forEach { types ->
+            findConverterIntoStatementInternal(
+                input = input,
+                columnTypes = types
+            )?.getOrCreateConverter()?.let {
+                return it
+            }
+        }
+        return null
+    }
+
+    private fun findConverterIntoStatementInternal(
+        input: XType,
+        columnTypes: List<XType>
+    ): TypeConverterEntry? {
+        if (columnTypes.isEmpty()) return null
+        val queue = TypeConverterQueue(
+            sourceType = input,
+            // each converter is keyed on which type they will take us to
+            keyType = TypeConverter::to
+        )
+
+        while (true) {
+            val current = queue.next() ?: break
+            val match = columnTypes.any { columnType ->
+                columnType.isSameType(current.type)
+            }
+            if (match) {
+                return current
+            }
+            // check for assignable matches but only enqueue them as there might be another shorter
+            // path
+            columnTypes.forEach { columnType ->
+                if (columnType.isAssignableFrom(current.type)) {
+                    queue.maybeEnqueue(
+                        current.appendConverter(
+                            UpCastTypeConverter(
+                                upCastFrom = current.type,
+                                upCastTo = columnType
+                            )
+                        )
+                    )
+                }
+            }
+            getAllTypeConvertersFrom(current.type).forEach {
+                queue.maybeEnqueue(current.appendConverter(it))
+            }
+        }
+        return null
+    }
+
+    override fun findConverterFromCursor(
+        columnTypes: List<XType>?,
+        output: XType
+    ): TypeConverter? {
+        @Suppress("NAME_SHADOWING") // intentional
+        val columnTypes = columnTypes ?: knownColumnTypes
+        // prefer nullable when reading from database, regardless of the output type
+        getColumnTypesInPreferenceBuckets(
+            nullability = NULLABLE,
+            explicitColumnTypes = columnTypes
+        ).forEach { types ->
+            findConverterFromCursorInternal(
+                columnTypes = types,
+                output = output
+            )?.let {
+                return it.getOrCreateConverter()
+            }
+        }
+
+        // if type is non-null, try to find nullable and add null check
+        return if (output.nullability == NONNULL) {
+            findConverterFromCursorInternal(
+                columnTypes = columnTypes,
+                output = output.makeNullable()
+            )?.appendConverter(
+                RequireNotNullTypeConverter(
+                    from = output.makeNullable()
+                )
+            )
+        } else {
+            null
+        }
+    }
+
+    private fun findConverterFromCursorInternal(
+        columnTypes: List<XType>,
+        output: XType
+    ): TypeConverterEntry? {
+        if (columnTypes.isEmpty()) return null
+        val queue = TypeConverterQueue(
+            sourceType = output,
+            // each converter is keyed on which type they receive as we are doing pathfinding
+            // reverse here
+            keyType = TypeConverter::from
+        )
+
+        while (true) {
+            val current = queue.next() ?: break
+            val match = columnTypes.any { columnType ->
+                columnType.isSameType(current.type)
+            }
+            if (match) {
+                return current
+            }
+            // check for assignable matches but only enqueue them as there might be another shorter
+            // path
+            columnTypes.forEach { columnType ->
+                if (current.type.isAssignableFrom(columnType)) {
+                    queue.maybeEnqueue(
+                        current.prependConverter(
+                            UpCastTypeConverter(
+                                upCastFrom = columnType,
+                                upCastTo = current.type
+                            )
+                        )
+                    )
+                }
+            }
+            getAllTypeConvertersTo(current.type).forEach {
+                queue.maybeEnqueue(current.prependConverter(it))
+            }
+        }
+        return null
+    }
+
+    override fun findTypeConverter(input: XType, output: XType): TypeConverter? {
+        return findConverterIntoStatementInternal(
+            input = input,
+            columnTypes = listOf(output)
+        )?.getOrCreateConverter()
+    }
+
+    /**
+     * Returns all type converters that can receive input type and return into another type.
+     */
+    private fun getAllTypeConvertersFrom(
+        input: XType
+    ): List<TypeConverter> {
+        // for input, check assignability because it defines whether we can use the method or not.
+        return typeConvertersByFromCache.getOrPut(input) {
+            // this cache avoids us many assignability checks.
+            typeConverters.mapNotNull { converter ->
+                when {
+                    converter.from.isSameType(input) -> converter
+                    converter.from.isAssignableFrom(input) -> CompositeTypeConverter(
+                        conv1 = UpCastTypeConverter(
+                            upCastFrom = input,
+                            upCastTo = converter.from
+                        ),
+                        conv2 = converter
+                    )
+                    else -> null
+                }
+            }
+        }
+    }
+
+    /**
+     * Returns all type converters that can return the output type.
+     */
+    private fun getAllTypeConvertersTo(
+        output: XType
+    ): List<TypeConverter> {
+        return typeConvertersByToCache.getOrPut(output) {
+            // this cache avoids us many assignability checks.
+            typeConverters.mapNotNull { converter ->
+                when {
+                    converter.to.isSameType(output) -> converter
+                    output.isAssignableFrom(converter.to) -> CompositeTypeConverter(
+                        conv1 = converter,
+                        conv2 = UpCastTypeConverter(
+                            upCastFrom = converter.to,
+                            upCastTo = output
+                        )
+                    )
+                    else -> null
+                }
+            }
+        }
+    }
+
+    /**
+     * Priority queue for the type converter search.
+     */
+    private class TypeConverterQueue(
+        sourceType: XType,
+        val keyType: TypeConverter.() -> XType
+    ) {
+        // using insertion order as the tie breaker for reproducible builds.
+        private var insertionOrder = 0
+
+        // map of XType to the converter that includes the path from the source type to the XType.
+        private val cheapestEntry = mutableMapOf<XType, TypeConverterEntry>()
+        private val queue = PriorityQueue<TypeConverterEntry>()
+
+        init {
+            val typeConverterEntry = TypeConverterEntry(
+                tieBreakerPriority = insertionOrder++,
+                type = sourceType,
+                converter = null
+            )
+            cheapestEntry[sourceType] = typeConverterEntry
+            queue.add(typeConverterEntry)
+        }
+
+        fun next(): TypeConverterEntry? {
+            while (queue.isNotEmpty()) {
+                val entry = queue.remove()
+                // check if we processed this type as there is no reason to process it again
+                if (cheapestEntry[entry.type] !== entry) {
+                    continue
+                }
+                return entry
+            }
+            return null
+        }
+
+        /**
+         * Enqueues the given [converter] if its target type (defined by [keyType]) is not visited
+         * or visited with a more expensive converter.
+         */
+        fun maybeEnqueue(
+            converter: TypeConverter
+        ): Boolean {
+            val keyType = converter.keyType()
+            val existing = cheapestEntry[keyType]
+            if (existing == null ||
+                (existing.converter != null && existing.converter.cost > converter.cost)
+            ) {
+                val entry = TypeConverterEntry(insertionOrder++, keyType, converter)
+                cheapestEntry[keyType] = entry
+                queue.add(entry)
+                return true
+            }
+            return false
+        }
+    }
+
+    private data class TypeConverterEntry(
+        // when costs are equal, tieBreakerPriority is used
+        val tieBreakerPriority: Int,
+        val type: XType,
+        val converter: TypeConverter?
+    ) : Comparable<TypeConverterEntry> {
+        override fun compareTo(other: TypeConverterEntry): Int {
+            if (converter == null) {
+                if (other.converter != null) {
+                    return -1
+                }
+            } else if (other.converter == null) {
+                return 1
+            } else {
+                val costCmp = converter.cost.compareTo(other.converter.cost)
+                if (costCmp != 0) {
+                    return costCmp
+                }
+            }
+            return tieBreakerPriority.compareTo(other.tieBreakerPriority)
+        }
+
+        fun getOrCreateConverter() = converter ?: NoOpConverter(type)
+
+        fun appendConverter(nextConverter: TypeConverter): TypeConverter {
+            if (converter == null) {
+                return nextConverter
+            }
+            return CompositeTypeConverter(
+                conv1 = converter,
+                conv2 = nextConverter
+            )
+        }
+
+        fun prependConverter(previous: TypeConverter): TypeConverter {
+            if (converter == null) {
+                return previous
+            }
+            return CompositeTypeConverter(
+                conv1 = previous,
+                conv2 = converter
+            )
+        }
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeAdapterStore.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeAdapterStore.kt
index 937ec76..d67bc34 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeAdapterStore.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeAdapterStore.kt
@@ -172,11 +172,14 @@
             ByteArrayColumnTypeAdapter.create(context.processingEnv).forEach(::addColumnAdapter)
             ByteBufferColumnTypeAdapter.create(context.processingEnv).forEach(::addColumnAdapter)
             PrimitiveBooleanToIntConverter.create(context.processingEnv).forEach(::addTypeConverter)
+            // null aware converter is able to automatically null wrap converters so we don't
+            // need this as long as we are running in KSP
             BoxedBooleanToBoxedIntConverter.create(context.processingEnv)
                 .forEach(::addTypeConverter)
             return TypeAdapterStore(
                 context = context, columnTypeAdapters = adapters,
-                typeConverterStore = TypeConverterStore(
+                typeConverterStore = TypeConverterStore.create(
+                    context = context,
                     typeConverters = converters,
                     knownColumnTypes = adapters.map { it.out }
                 ),
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStore.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStore.kt
index fbd8863..0b7d0af 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStore.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStore.kt
@@ -17,49 +17,34 @@
 package androidx.room.solver
 
 import androidx.room.compiler.processing.XType
+import androidx.room.processor.Context
 import androidx.room.solver.types.CompositeTypeConverter
 import androidx.room.solver.types.NoOpConverter
+import androidx.room.solver.types.RequireNotNullTypeConverter
 import androidx.room.solver.types.TypeConverter
-import java.util.LinkedList
+import androidx.room.solver.types.UpCastTypeConverter
 
-/**
- * Common logic that handles conversion between types either using built in converters or user
- * provided type converters.
- */
-class TypeConverterStore(
-    /**
-     * Available TypeConverters
-     */
-    private val typeConverters: List<TypeConverter>,
-    /**
-     * List of types that can be saved into db/read from without a converter.
-     */
-    private val knownColumnTypes: List<XType>
-) {
+interface TypeConverterStore {
+    val typeConverters: List<TypeConverter>
+
     /**
      * Finds a [TypeConverter] (might be composite) that can convert the given [input] type into
      * one of the given [columnTypes]. If [columnTypes] is not specified, targets all
-     * [knownColumnTypes].
+     * `knownColumnTypes`.
      */
     fun findConverterIntoStatement(
         input: XType,
-        columnTypes: List<XType>? = null
-    ) = findTypeConverter(
-        inputs = listOf(input),
-        outputs = columnTypes ?: knownColumnTypes
-    )
+        columnTypes: List<XType>?
+    ): TypeConverter?
 
     /**
      * Finds a [TypeConverter] (might be composite) that can convert the given [columnTypes] into
-     * the [output] type. If [columnTypes] is not specified, uses all [knownColumnTypes].
+     * the [output] type. If [columnTypes] is not specified, uses all `knownColumnTypes`.
      */
     fun findConverterFromCursor(
-        columnTypes: List<XType>? = null,
+        columnTypes: List<XType>?,
         output: XType
-    ) = findTypeConverter(
-        inputs = columnTypes ?: knownColumnTypes,
-        outputs = listOf(output)
-    )
+    ): TypeConverter?
 
     /**
      * Finds a [TypeConverter] from [input] to [output].
@@ -67,104 +52,8 @@
     fun findTypeConverter(
         input: XType,
         output: XType
-    ) = findTypeConverter(
-        inputs = listOf(input),
-        outputs = listOf(output)
-    )
+    ): TypeConverter?
 
-    /**
-     * Finds a type converter that can convert one of the input values to one of the output values.
-     *
-     * When multiple conversion paths are possible, shortest path (least amount of conversion) is
-     * preferred.
-     */
-    private fun findTypeConverter(
-        inputs: List<XType>,
-        outputs: List<XType>
-    ): TypeConverter? {
-        if (inputs.isEmpty()) {
-            return null
-        }
-        inputs.forEach { input ->
-            if (outputs.any { output -> input.isSameType(output) }) {
-                return NoOpConverter(input)
-            }
-        }
-
-        val excludes = arrayListOf<XType>()
-
-        val queue = LinkedList<TypeConverter>()
-        fun List<TypeConverter>.findMatchingConverter(): TypeConverter? {
-            // We prioritize exact match over assignable. To do that, this variable keeps any
-            // assignable match and if we cannot find exactly same type match, we'll return the
-            // assignable match.
-            var assignableMatchFallback: TypeConverter? = null
-            this.forEach { converter ->
-                outputs.forEach { output ->
-                    if (output.isSameType(converter.to)) {
-                        return converter
-                    } else if (assignableMatchFallback == null &&
-                        output.isAssignableFrom(converter.to)
-                    ) {
-                        // if we don't find exact match, we'll return this.
-                        assignableMatchFallback = converter
-                    }
-                }
-            }
-            return assignableMatchFallback
-        }
-        inputs.forEach { input ->
-            val candidates = getAllTypeConverters(input, excludes)
-            val match = candidates.findMatchingConverter()
-            if (match != null) {
-                return match
-            }
-            candidates.forEach {
-                excludes.add(it.to)
-                queue.add(it)
-            }
-        }
-        excludes.addAll(inputs)
-        while (queue.isNotEmpty()) {
-            val prev = queue.pop()
-            val from = prev.to
-            val candidates = getAllTypeConverters(from, excludes)
-            val match = candidates.findMatchingConverter()
-            if (match != null) {
-                return CompositeTypeConverter(prev, match)
-            }
-            candidates.forEach {
-                excludes.add(it.to)
-                queue.add(CompositeTypeConverter(prev, it))
-            }
-        }
-        return null
-    }
-
-    /**
-     * Returns all type converters that can receive input type and return into another type.
-     * The returned list is ordered by priority such that if we have an exact match, it is
-     * prioritized.
-     */
-    private fun getAllTypeConverters(input: XType, excludes: List<XType>): List<TypeConverter> {
-        // for input, check assignability because it defines whether we can use the method or not.
-        // for excludes, use exact match
-        return typeConverters.filter { converter ->
-            converter.from.isAssignableFrom(input) &&
-                !excludes.any { it.isSameType(converter.to) }
-        }.sortedByDescending {
-            // if it is the same, prioritize
-            if (it.from.isSameType(input)) {
-                2
-            } else {
-                1
-            }
-        }
-    }
-
-    /**
-     * Tries to reverse the converter going through the same nodes, if possible.
-     */
     fun reverse(converter: TypeConverter): TypeConverter? {
         return when (converter) {
             is NoOpConverter -> converter
@@ -173,6 +62,11 @@
                 val r2 = reverse(converter.conv2) ?: return null
                 CompositeTypeConverter(r2, r1)
             }
+            // reverse of require not null is upcast since not null can be converted into nullable
+            is RequireNotNullTypeConverter -> UpCastTypeConverter(
+                upCastFrom = converter.to,
+                upCastTo = converter.from
+            )
             else -> {
                 typeConverters.firstOrNull {
                     it.from.isSameType(converter.to) &&
@@ -181,4 +75,30 @@
             }
         }
     }
+
+    companion object {
+        /**
+         * @param context Processing context
+         * @param typeConverters Available TypeConverters, ordered by priority when they have the
+         *        same cost.
+         * @param knownColumnTypes List of types that can be saved into db/read from without a
+         *        converter.
+         */
+        fun create(
+            context: Context,
+            typeConverters: List<TypeConverter>,
+            knownColumnTypes: List<XType>
+        ) = if (context.useNullAwareConverter) {
+            NullAwareTypeConverterStore(
+                context = context,
+                typeConverters = typeConverters,
+                knownColumnTypes = knownColumnTypes
+            )
+        } else {
+            TypeConverterStoreImpl(
+                typeConverters = typeConverters,
+                knownColumnTypes = knownColumnTypes
+            )
+        }
+    }
 }
\ No newline at end of file
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStoreImpl.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStoreImpl.kt
new file mode 100644
index 0000000..848addf
--- /dev/null
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/TypeConverterStoreImpl.kt
@@ -0,0 +1,152 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver
+
+import androidx.room.compiler.processing.XType
+import androidx.room.solver.types.CompositeTypeConverter
+import androidx.room.solver.types.NoOpConverter
+import androidx.room.solver.types.TypeConverter
+import java.util.LinkedList
+
+/**
+ * Legacy [TypeConverterStore] implementation that does not assume we have type nullability
+ * information. It is kept around for backwards compatibility.
+ */
+class TypeConverterStoreImpl(
+    /**
+     * Available TypeConverters
+     */
+    override val typeConverters: List<TypeConverter>,
+    /**
+     * List of types that can be saved into db/read from without a converter.
+     */
+    private val knownColumnTypes: List<XType>
+) : TypeConverterStore {
+    override fun findConverterIntoStatement(
+        input: XType,
+        columnTypes: List<XType>?
+    ) = findTypeConverter(
+        inputs = listOf(input),
+        outputs = columnTypes ?: knownColumnTypes
+    )
+
+    override fun findConverterFromCursor(
+        columnTypes: List<XType>?,
+        output: XType
+    ) = findTypeConverter(
+        inputs = columnTypes ?: knownColumnTypes,
+        outputs = listOf(output)
+    )
+
+    override fun findTypeConverter(
+        input: XType,
+        output: XType
+    ) = findTypeConverter(
+        inputs = listOf(input),
+        outputs = listOf(output)
+    )
+
+    /**
+     * Finds a type converter that can convert one of the input values to one of the output values.
+     *
+     * When multiple conversion paths are possible, shortest path (least amount of conversion) is
+     * preferred.
+     */
+    private fun findTypeConverter(
+        inputs: List<XType>,
+        outputs: List<XType>
+    ): TypeConverter? {
+        if (inputs.isEmpty()) {
+            return null
+        }
+        inputs.forEach { input ->
+            if (outputs.any { output -> input.isSameType(output) }) {
+                return NoOpConverter(input)
+            }
+        }
+
+        val excludes = arrayListOf<XType>()
+
+        val queue = LinkedList<TypeConverter>()
+        fun List<TypeConverter>.findMatchingConverter(): TypeConverter? {
+            // We prioritize exact match over assignable. To do that, this variable keeps any
+            // assignable match and if we cannot find exactly same type match, we'll return the
+            // assignable match.
+            var assignableMatchFallback: TypeConverter? = null
+            this.forEach { converter ->
+                outputs.forEach { output ->
+                    if (output.isSameType(converter.to)) {
+                        return converter
+                    } else if (assignableMatchFallback == null &&
+                        output.isAssignableFrom(converter.to)
+                    ) {
+                        // if we don't find exact match, we'll return this.
+                        assignableMatchFallback = converter
+                    }
+                }
+            }
+            return assignableMatchFallback
+        }
+        inputs.forEach { input ->
+            val candidates = getAllTypeConverters(input, excludes)
+            val match = candidates.findMatchingConverter()
+            if (match != null) {
+                return match
+            }
+            candidates.forEach {
+                excludes.add(it.to)
+                queue.add(it)
+            }
+        }
+        excludes.addAll(inputs)
+        while (queue.isNotEmpty()) {
+            val prev = queue.pop()
+            val from = prev.to
+            val candidates = getAllTypeConverters(from, excludes)
+            val match = candidates.findMatchingConverter()
+            if (match != null) {
+                return CompositeTypeConverter(prev, match)
+            }
+            candidates.forEach {
+                excludes.add(it.to)
+                queue.add(CompositeTypeConverter(prev, it))
+            }
+        }
+        return null
+    }
+
+    /**
+     * Returns all type converters that can receive input type and return into another type.
+     * The returned list is ordered by priority such that if we have an exact match, it is
+     * prioritized.
+     */
+    private fun getAllTypeConverters(input: XType, excludes: List<XType>): List<TypeConverter> {
+        // for input, check assignability because it defines whether we can use the method or not.
+        // for excludes, use exact match
+        return typeConverters.filter { converter ->
+            converter.from.isAssignableFrom(input) &&
+                !excludes.any { it.isSameType(converter.to) }
+        }.sortedByDescending {
+            // if it is the same, prioritize
+            if (it.from.isSameType(input)) {
+                2
+            } else {
+                1
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/CompositeTypeConverter.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/CompositeTypeConverter.kt
index b9c7cb9..2bb2c0c 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/CompositeTypeConverter.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/CompositeTypeConverter.kt
@@ -22,7 +22,9 @@
  * combines 2 type converters
  */
 class CompositeTypeConverter(val conv1: TypeConverter, val conv2: TypeConverter) : TypeConverter(
-    conv1.from, conv2.to
+    from = conv1.from,
+    to = conv2.to,
+    cost = conv1.cost + conv2.cost
 ) {
     override fun doConvert(inputVarName: String, outputVarName: String, scope: CodeGenScope) {
         scope.builder().apply {
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/NullAwareTypeConverters.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/NullAwareTypeConverters.kt
new file mode 100644
index 0000000..288899f
--- /dev/null
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/NullAwareTypeConverters.kt
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver.types
+
+import androidx.annotation.VisibleForTesting
+import androidx.room.compiler.processing.XNullability
+import androidx.room.compiler.processing.XType
+import androidx.room.ext.L
+import androidx.room.ext.S
+import androidx.room.ext.T
+import androidx.room.solver.CodeGenScope
+import java.lang.IllegalStateException
+
+/**
+ * A type converter that checks if the input is null and returns null instead of calling the
+ * [delegate].
+ */
+class NullSafeTypeConverter(
+    @VisibleForTesting
+    internal val delegate: TypeConverter
+) : TypeConverter(
+    from = delegate.from.makeNullable(),
+    to = delegate.to.makeNullable(),
+    cost = delegate.cost + Cost.NULL_SAFE
+) {
+    init {
+        check(delegate.from.nullability == XNullability.NONNULL) {
+            "NullableWrapper can ony be used if the input type is non-nullable"
+        }
+    }
+
+    override fun doConvert(inputVarName: String, outputVarName: String, scope: CodeGenScope) {
+        scope.builder().apply {
+            beginControlFlow("if($L == null)", inputVarName)
+            addStatement("$L = null", outputVarName)
+            nextControlFlow("else")
+            delegate.convert(inputVarName, outputVarName, scope)
+            endControlFlow()
+        }
+    }
+}
+
+/**
+ * A [TypeConverter] that checks the value is `non-null` and throws if it is null.
+ */
+class RequireNotNullTypeConverter(
+    from: XType,
+) : TypeConverter(
+    from = from,
+    to = from.makeNonNullable(),
+    cost = Cost.REQUIRE_NOT_NULL
+) {
+    init {
+        check(from.nullability != XNullability.NONNULL) {
+            "No reason to null check a non-null input"
+        }
+    }
+
+    override fun doConvert(inputVarName: String, outputVarName: String, scope: CodeGenScope) {
+        scope.builder().apply {
+            beginControlFlow("if($L == null)", inputVarName)
+            addStatement(
+                "throw new $T($S)", IllegalStateException::class.java,
+                "Expected non-null ${from.typeName}, but it was null."
+            )
+            nextControlFlow("else").apply {
+                addStatement("$L = $L", outputVarName, inputVarName)
+            }
+            endControlFlow()
+        }
+    }
+
+    override fun doConvert(inputVarName: String, scope: CodeGenScope): String {
+        scope.builder().apply {
+            beginControlFlow("if($L == null)", inputVarName)
+            addStatement(
+                "throw new $T($S)", IllegalStateException::class.java,
+                "Expected non-null ${from.typeName}, but it was null."
+            )
+            endControlFlow()
+        }
+        return inputVarName
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/TypeConverter.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/TypeConverter.kt
index fab9eb1..2f8ae17 100644
--- a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/TypeConverter.kt
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/TypeConverter.kt
@@ -16,6 +16,7 @@
 
 package androidx.room.solver.types
 
+import androidx.annotation.VisibleForTesting
 import androidx.room.compiler.processing.XType
 import androidx.room.ext.L
 import androidx.room.ext.T
@@ -24,7 +25,11 @@
 /**
  * A code generator that can convert from 1 type to another
  */
-abstract class TypeConverter(val from: XType, val to: XType) {
+abstract class TypeConverter(
+    val from: XType,
+    val to: XType,
+    val cost: Cost = Cost.CONVERTER
+) {
     /**
      * Should generate the code that will covert [inputVarName] of type [from] to [outputVarName]
      * of type [to]. This method *should not* declare the [outputVarName] as it is already
@@ -70,4 +75,123 @@
     ) {
         doConvert(inputVarName, outputVarName, scope)
     }
+
+    /**
+     * Represents the cost of a type converter.
+     *
+     * When calculating cost, we consider multiple types of conversions in ascending order, from
+     * cheapest to expensive:
+     * * `upcast`: Converts from a subtype to super type (e.g. Int to Number)
+     * * `nullSafeWrapper`: Adds a null check before calling the delegated converter or else
+     *    returns null.
+     * * `converter`: Unit converter
+     * * `requireNotNull`: Adds a null check before returning the delegated converter's value
+     *    or throws if the value is null.
+     *
+     * The comparison happens in buckets such that having 10 upcasts is still cheaper than having
+     * 1 nullSafeWrapper.
+     *
+     * Internally, this class uses an IntArray to keep its fields to optimize for readability in
+     * operators.
+     */
+    class Cost private constructor(
+        /**
+         * Values for each bucket, ordered from most expensive to least expensive.
+         */
+        private val values: IntArray
+    ) : Comparable<Cost> {
+        init {
+            require(values.size == Buckets.SIZE)
+        }
+
+        constructor(
+            converters: Int,
+            nullSafeWrapper: Int = 0,
+            upCasts: Int = 0,
+            requireNotNull: Int = 0
+        ) : this(
+            // NOTE: construction order here MUST match the [Buckets]
+            intArrayOf(
+                requireNotNull,
+                converters,
+                nullSafeWrapper,
+                upCasts
+            )
+        )
+
+        @VisibleForTesting
+        val upCasts: Int
+            get() = values[Buckets.UP_CAST]
+
+        @VisibleForTesting
+        val nullSafeWrapper: Int
+            get() = values[Buckets.NULL_SAFE]
+
+        @VisibleForTesting
+        val requireNotNull: Int
+            get() = values[Buckets.REQUIRE_NOT_NULL]
+
+        @VisibleForTesting
+        val converters: Int
+            get() = values[Buckets.CONVERTER]
+
+        operator fun plus(other: Cost) = Cost(
+            values = IntArray(Buckets.SIZE) { index ->
+                values[index] + other.values[index]
+            }
+        )
+
+        override operator fun compareTo(other: Cost): Int {
+            for (index in 0 until Buckets.SIZE) {
+                val cmp = values[index].compareTo(other.values[index])
+                if (cmp != 0) {
+                    return cmp
+                }
+            }
+            return 0
+        }
+
+        override fun toString() = buildString {
+            append("Cost[")
+            append("upcast:")
+            append(upCasts)
+            append(",nullsafe:")
+            append(nullSafeWrapper)
+            append(",converters:")
+            append(converters)
+            append(",requireNotNull:")
+            append(requireNotNull)
+            append("]")
+        }
+
+        override fun equals(other: Any?): Boolean {
+            if (other !is Cost) {
+                return false
+            }
+            return compareTo(other) == 0
+        }
+
+        override fun hashCode(): Int {
+            // we don't really use hash functions so this is good enough as a hash function.
+            return values[Buckets.CONVERTER]
+        }
+
+        companion object {
+            val UP_CAST = Cost(converters = 0, upCasts = 1)
+            val NULL_SAFE = Cost(converters = 0, nullSafeWrapper = 1)
+            val CONVERTER = Cost(converters = 1)
+            val REQUIRE_NOT_NULL = Cost(converters = 0, requireNotNull = 1)
+        }
+
+        /**
+         * Comparison buckets, ordered from the MOST expensive to LEAST expensive
+         */
+        private object Buckets {
+            const val REQUIRE_NOT_NULL = 0
+            const val CONVERTER = 1
+            const val NULL_SAFE = 2
+            const val UP_CAST = 3
+            const val SIZE = 4
+        }
+    }
 }
diff --git a/room/room-compiler/src/main/kotlin/androidx/room/solver/types/UpCastTypeConverter.kt b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/UpCastTypeConverter.kt
new file mode 100644
index 0000000..ad07028
--- /dev/null
+++ b/room/room-compiler/src/main/kotlin/androidx/room/solver/types/UpCastTypeConverter.kt
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver.types
+
+import androidx.room.compiler.processing.XType
+import androidx.room.ext.L
+import androidx.room.solver.CodeGenScope
+
+/**
+ * We have a special class for upcasting types in type converters. (e.g. Int to Number)
+ * It is used in the pathfinding to be more expensive than exactly matching calls to prioritize
+ * exact matches.
+ */
+class UpCastTypeConverter(
+    upCastFrom: XType,
+    upCastTo: XType
+) : TypeConverter(
+    from = upCastFrom,
+    to = upCastTo,
+    cost = Cost.UP_CAST
+) {
+    override fun doConvert(inputVarName: String, outputVarName: String, scope: CodeGenScope) {
+        scope.builder().apply {
+            addStatement("$L = $L", outputVarName, inputVarName)
+        }
+    }
+
+    override fun doConvert(inputVarName: String, scope: CodeGenScope): String {
+        // normally, we don't need to generate any code here but if the upcast is converting from
+        // a primitive to boxed; we need to. Otherwise, output value won't become an object and
+        // that might break the rest of the code generation (e.g. checking nullable on primitive)
+        return if (to.typeName.isBoxedPrimitive && from.typeName.isPrimitive) {
+            super.doConvert(inputVarName, scope)
+        } else {
+            inputVarName
+        }
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/test/kotlin/androidx/room/solver/NullabilityAwareTypeConverterStoreTest.kt b/room/room-compiler/src/test/kotlin/androidx/room/solver/NullabilityAwareTypeConverterStoreTest.kt
new file mode 100644
index 0000000..99a0162
--- /dev/null
+++ b/room/room-compiler/src/test/kotlin/androidx/room/solver/NullabilityAwareTypeConverterStoreTest.kt
@@ -0,0 +1,743 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver
+
+import androidx.room.RoomKspProcessor
+import androidx.room.compiler.processing.XType
+import androidx.room.compiler.processing.util.Source
+import androidx.room.compiler.processing.util.XTestInvocation
+import androidx.room.compiler.processing.util.compiler.TestCompilationArguments
+import androidx.room.compiler.processing.util.compiler.compile
+import androidx.room.compiler.processing.util.runProcessorTest
+import androidx.room.processor.Context.BooleanProcessorOptions.USE_NULL_AWARE_CONVERTER
+import androidx.room.processor.CustomConverterProcessor
+import androidx.room.processor.DaoProcessor
+import androidx.room.solver.types.CustomTypeConverterWrapper
+import androidx.room.solver.types.TypeConverter
+import androidx.room.testing.context
+import androidx.room.vo.BuiltInConverterFlags
+import androidx.room.writer.DaoWriter
+import com.google.common.truth.Truth.assertThat
+import org.junit.Rule
+import org.junit.Test
+import org.junit.rules.TemporaryFolder
+import org.junit.runner.RunWith
+import org.junit.runners.JUnit4
+import javax.tools.Diagnostic
+
+@RunWith(JUnit4::class)
+class NullabilityAwareTypeConverterStoreTest {
+    @get:Rule
+    val tmpFolder = TemporaryFolder()
+    val source = Source.kotlin(
+        "Foo.kt",
+        """
+            import androidx.room.*
+            class MyClass
+            class NonNullConverters {
+                @TypeConverter
+                fun myClassToString(myClass: MyClass): String {
+                    TODO()
+                }
+                @TypeConverter
+                fun stringToMyClass(input: String): MyClass {
+                    TODO()
+                }
+            }
+            class MyNullableReceivingConverters {
+                @TypeConverter
+                fun nullableMyClassToNonNullString(myClass: MyClass?): String {
+                    TODO()
+                }
+                @TypeConverter
+                fun nullableStringToNonNullMyClass(input: String?): MyClass {
+                    TODO()
+                }
+            }
+            class MyFullyNullableConverters {
+                @TypeConverter
+                fun nullableMyClassToNullableString(myClass: MyClass?): String? {
+                    TODO()
+                }
+                @TypeConverter
+                fun nullableStringToNullableMyClass(input: String?): MyClass? {
+                    TODO()
+                }
+            }
+        """.trimIndent()
+    )
+
+    private fun XTestInvocation.createStore(
+        vararg converters: String
+    ): TypeConverterStore {
+        val allConverters = converters.flatMap {
+            CustomConverterProcessor(
+                context = context,
+                element = processingEnv.requireTypeElement(it)
+            ).process()
+        }.map(::CustomTypeConverterWrapper)
+        return TypeAdapterStore.create(
+            context = context,
+            builtInConverterFlags = BuiltInConverterFlags.DEFAULT,
+            allConverters
+        ).typeConverterStore
+    }
+
+    @Test
+    fun withNonNullableConverters() {
+        val result = collectStringConversionResults("NonNullConverters")
+        assertResult(
+            result.trim(),
+            """
+            JAVAC
+            String? to MyClass?: stringToMyClass
+            MyClass? to String?: myClassToString
+            String? to MyClass!: stringToMyClass
+            MyClass! to String?: myClassToString
+            String! to MyClass?: stringToMyClass
+            MyClass? to String!: myClassToString
+            String! to MyClass!: stringToMyClass
+            MyClass! to String!: myClassToString
+            KSP
+            String? to MyClass?: (String? == null ? null : stringToMyClass)
+            MyClass? to String?: (MyClass? == null ? null : myClassToString)
+            String? to MyClass!: null
+            MyClass! to String?: myClassToString / (String! as String?)
+            String! to MyClass?: stringToMyClass / (MyClass! as MyClass?)
+            MyClass? to String!: null
+            String! to MyClass!: stringToMyClass
+            MyClass! to String!: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun withNonNullableConverters_cursor() {
+        val result = collectCursorResults("NonNullConverters")
+        assertResult(
+            result.trim(),
+            """
+                JAVAC
+                Cursor to MyClass?: stringToMyClass
+                MyClass? to Cursor: myClassToString
+                Cursor to MyClass!: stringToMyClass
+                MyClass! to Cursor: myClassToString
+                KSP
+                Cursor to MyClass?: (String? == null ? null : stringToMyClass)
+                MyClass? to Cursor: (MyClass? == null ? null : myClassToString)
+                // when reading from cursor, we can assume non-null cursor value when
+                // we don't have a converter that would convert it from String?
+                Cursor to MyClass!: stringToMyClass
+                MyClass! to Cursor: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun withNonNullAndNullableReceiving() {
+        val result = collectStringConversionResults(
+            "NonNullConverters",
+            "MyNullableReceivingConverters"
+        )
+        assertResult(
+            result.trim(),
+            """
+            JAVAC
+            String? to MyClass?: stringToMyClass
+            MyClass? to String?: myClassToString
+            String? to MyClass!: stringToMyClass
+            MyClass! to String?: myClassToString
+            String! to MyClass?: stringToMyClass
+            MyClass? to String!: myClassToString
+            String! to MyClass!: stringToMyClass
+            MyClass! to String!: myClassToString
+            KSP
+            String? to MyClass?: nullableStringToNonNullMyClass / (MyClass! as MyClass?)
+            MyClass? to String?: nullableMyClassToNonNullString / (String! as String?)
+            String? to MyClass!: nullableStringToNonNullMyClass
+            MyClass! to String?: myClassToString / (String! as String?)
+            String! to MyClass?: stringToMyClass / (MyClass! as MyClass?)
+            MyClass? to String!: nullableMyClassToNonNullString
+            String! to MyClass!: stringToMyClass
+            MyClass! to String!: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun withNonNullAndNullableReceiving_cursor() {
+        val result = collectCursorResults(
+            "NonNullConverters",
+            "MyNullableReceivingConverters"
+        )
+        assertResult(
+            result.trim(),
+            """
+            JAVAC
+            Cursor to MyClass?: stringToMyClass
+            MyClass? to Cursor: myClassToString
+            Cursor to MyClass!: stringToMyClass
+            MyClass! to Cursor: myClassToString
+            KSP
+            // we start from nullable string because cursor values are assumed nullable when reading
+            Cursor to MyClass?: nullableStringToNonNullMyClass / (MyClass! as MyClass?)
+            // there is an additional upcast for String! to String? because when the written value
+            // is nullable, we prioritize a nullable column
+            MyClass? to Cursor: nullableMyClassToNonNullString / (String! as String?)
+            Cursor to MyClass!: nullableStringToNonNullMyClass
+            MyClass! to Cursor: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun withFullyNullableConverters() {
+        val result = collectStringConversionResults(
+            "NonNullConverters",
+            "MyNullableReceivingConverters",
+            "MyFullyNullableConverters"
+        )
+        assertResult(
+            result.trim(),
+            """
+                JAVAC
+                String? to MyClass?: stringToMyClass
+                MyClass? to String?: myClassToString
+                String? to MyClass!: stringToMyClass
+                MyClass! to String?: myClassToString
+                String! to MyClass?: stringToMyClass
+                MyClass? to String!: myClassToString
+                String! to MyClass!: stringToMyClass
+                MyClass! to String!: myClassToString
+                KSP
+                String? to MyClass?: nullableStringToNullableMyClass
+                MyClass? to String?: nullableMyClassToNullableString
+                String? to MyClass!: nullableStringToNonNullMyClass
+                // another alternative is to use nonNullMyClassToNullableString and then upcast
+                // both are equal weight
+                MyClass! to String?: (MyClass! as MyClass?) / nullableMyClassToNullableString
+                String! to MyClass?: (String! as String?) / nullableStringToNullableMyClass
+                MyClass? to String!: nullableMyClassToNonNullString
+                String! to MyClass!: stringToMyClass
+                MyClass! to String!: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun withFullyNullableConverters_cursor() {
+        val result = collectCursorResults(
+            "NonNullConverters",
+            "MyNullableReceivingConverters",
+            "MyFullyNullableConverters"
+        )
+        assertResult(
+            result.trim(),
+            """
+                JAVAC
+                Cursor to MyClass?: stringToMyClass
+                MyClass? to Cursor: myClassToString
+                Cursor to MyClass!: stringToMyClass
+                MyClass! to Cursor: myClassToString
+                KSP
+                Cursor to MyClass?: nullableStringToNullableMyClass
+                MyClass? to Cursor: nullableMyClassToNullableString
+                Cursor to MyClass!: nullableStringToNonNullMyClass
+                MyClass! to Cursor: myClassToString
+            """.trimIndent()
+        )
+    }
+
+    @Test
+    fun pojoProcess() {
+        // This is a repro case from trying to run TestApp with null aware converter.
+        // It reproduces the case where if we don't know nullability, we shouldn't try to
+        // prioritize nullable or non-null; instead YOLO and find whichever we can find first.
+        val user = Source.java(
+            "User", """
+            import androidx.room.*;
+            import java.util.*;
+            @TypeConverters({TestConverters.class})
+            @Entity
+            public class User {
+                @PrimaryKey
+                public int mId;
+                public Set<Day> mWorkDays = new HashSet<>();
+            }
+        """.trimIndent()
+        )
+        val converters = Source.java(
+            "TestConverters", """
+            import androidx.room.*;
+            import java.util.Date;
+            import java.util.HashSet;
+            import java.util.Set;
+            class TestConverters {
+                @TypeConverter
+                public static Set<Day> decomposeDays(int flags) {
+                    Set<Day> result = new HashSet<>();
+                    for (Day day : Day.values()) {
+                        if ((flags & (1 << day.ordinal())) != 0) {
+                            result.add(day);
+                        }
+                    }
+                    return result;
+                }
+
+                @TypeConverter
+                public static int composeDays(Set<Day> days) {
+                    int result = 0;
+                    for (Day day : days) {
+                        result |= 1 << day.ordinal();
+                    }
+                    return result;
+                }
+            }
+        """.trimIndent()
+        )
+        val day = Source.java(
+            "Day", """
+            public enum Day {
+                MONDAY,
+                TUESDAY,
+                WEDNESDAY,
+                THURSDAY,
+                FRIDAY,
+                SATURDAY,
+                SUNDAY
+            }
+        """.trimIndent()
+        )
+        val dao = Source.java(
+            "MyDao", """
+            import androidx.room.*;
+            @Dao
+            interface MyDao {
+                @Insert
+                void insert(User user);
+            }
+        """.trimIndent()
+        )
+        runProcessorTest(
+            sources = listOf(user, day, converters, dao),
+            options = mapOf(
+                USE_NULL_AWARE_CONVERTER.argName to "true"
+            )
+        ) { invocation ->
+            val daoProcessor = DaoProcessor(
+                baseContext = invocation.context,
+                element = invocation.processingEnv.requireTypeElement("MyDao"),
+                dbType = invocation.processingEnv.requireType("androidx.room.RoomDatabase"),
+                dbVerifier = null
+            )
+            DaoWriter(
+                dao = daoProcessor.process(),
+                dbElement = invocation.processingEnv
+                    .requireTypeElement("androidx.room.RoomDatabase"),
+                processingEnv = invocation.processingEnv
+            ).write(invocation.processingEnv)
+            invocation.assertCompilationResult {
+                generatedSourceFileWithPath("MyDao_Impl.java").let {
+                    // make sure it bounded w/o upcasting to Boolean
+                    it.contains("final int _tmp = TestConverters.composeDays(value.mWorkDays);")
+                    it.contains("stmt.bindLong(2, _tmp);")
+                }
+            }
+        }
+    }
+
+    @Test
+    fun checkSyntheticConverters() {
+        class MockTypeConverter(
+            from: XType,
+            to: XType,
+        ) : TypeConverter(
+            from = from,
+            to = to
+        ) {
+            override fun doConvert(
+                inputVarName: String,
+                outputVarName: String,
+                scope: CodeGenScope
+            ) {
+            }
+        }
+        runProcessorTest { invocation ->
+            val string = invocation.processingEnv.requireType(String::class)
+                .makeNonNullable()
+            val int = invocation.processingEnv.requireType(Int::class)
+                .makeNonNullable()
+            val long = invocation.processingEnv.requireType(Long::class)
+                .makeNonNullable()
+            val number = invocation.processingEnv.requireType(Number::class)
+                .makeNonNullable()
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int.makeNullable()
+                    )
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // nullable converter, don't duplicate anything
+                assertThat(
+                    store.typeConverters
+                ).hasSize(1)
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = int
+                    )
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                if (invocation.isKsp) {
+                    // add a null wrapper version
+                    assertThat(store.typeConverters).hasSize(2)
+                } else {
+                    // do not duplicate unless we run in KSP
+                    assertThat(store.typeConverters).hasSize(1)
+                }
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = int
+                    ),
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int
+                    )
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // don't duplicate, we already have a null receiving version
+                assertThat(store.typeConverters).hasSize(2)
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = int
+                    ),
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int.makeNullable()
+                    )
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // don't duplicate, we already have a null receiving version
+                assertThat(store.typeConverters).hasSize(2)
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = int
+                    ),
+                    MockTypeConverter(
+                        from = string,
+                        to = long
+                    ),
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int.makeNullable()
+                    )
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // don't duplicate, we already have a null receiving version
+                if (invocation.isKsp) {
+                    // duplicate the long receiving one
+                    assertThat(store.typeConverters).hasSize(4)
+                } else {
+                    // don't duplicate in javac
+                    assertThat(store.typeConverters).hasSize(3)
+                }
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = number
+                    ),
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int
+                    ),
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // don't duplicate string number converter since we have string? to int
+                assertThat(store.typeConverters).hasSize(2)
+            }
+            NullAwareTypeConverterStore(
+                context = invocation.context,
+                typeConverters = listOf(
+                    MockTypeConverter(
+                        from = string,
+                        to = number.makeNullable()
+                    ),
+                    MockTypeConverter(
+                        from = string.makeNullable(),
+                        to = int
+                    ),
+                ),
+                knownColumnTypes = emptyList()
+            ).let { store ->
+                // don't duplicate string number converter since we have string? to int
+                assertThat(store.typeConverters).hasSize(2)
+            }
+        }
+    }
+
+    @Test
+    fun warnIfTurnedOffInKsp() {
+        val sources = Source.kotlin("Foo.kt", "")
+        arrayOf("", "true", "false").forEach { value ->
+            val result = compile(
+                workingDir = tmpFolder.newFolder(),
+                arguments = TestCompilationArguments(
+                    sources = listOf(sources),
+                    symbolProcessorProviders = listOf(
+                        RoomKspProcessor.Provider()
+                    ),
+                    processorOptions = mapOf(
+                        USE_NULL_AWARE_CONVERTER.argName to value
+                    )
+                )
+            )
+            val warnings = result.diagnostics[Diagnostic.Kind.WARNING]?.map {
+                it.msg
+            }?.filter {
+                it.contains("Disabling null-aware type analysis in KSP is a temporary flag")
+            } ?: emptyList()
+            val expected = if (value == "false") {
+                1
+            } else {
+                0
+            }
+            assertThat(
+                warnings
+            ).hasSize(expected)
+        }
+    }
+
+    /**
+     * Test converting a known column type into another type due to explicit affinity
+     */
+    @Test
+    fun knownColumnTypeToExplicitType() {
+        val source = Source.kotlin(
+            "Subject.kt", """
+            import androidx.room.*
+            object MyByteArrayConverter {
+                @TypeConverter
+                fun toByteArray(input:String): ByteArray { TODO() }
+                @TypeConverter
+                fun fromByteArray(input:ByteArray): String { TODO() }
+            }
+            class Subject(val arr:ByteArray)
+        """.trimIndent()
+        )
+        runProcessorTest(
+            sources = listOf(source),
+            options = mapOf(
+                USE_NULL_AWARE_CONVERTER.argName to "true"
+            )
+        ) { invocation ->
+            val byteArray = invocation.processingEnv.requireTypeElement("Subject")
+                .getDeclaredFields().first().type.makeNonNullable()
+            val string = invocation.processingEnv.requireType("java.lang.String")
+            invocation.createStore().let { storeWithoutConverter ->
+                val intoStatement = storeWithoutConverter.findConverterIntoStatement(
+                    input = byteArray,
+                    columnTypes = listOf(
+                        string.makeNullable(),
+                        string.makeNonNullable()
+                    )
+                )
+                assertThat(intoStatement).isNull()
+                val fromCursor = storeWithoutConverter.findConverterFromCursor(
+                    output = byteArray,
+                    columnTypes = listOf(
+                        string.makeNullable(),
+                        string.makeNonNullable()
+                    )
+                )
+                assertThat(fromCursor).isNull()
+            }
+
+            invocation.createStore(
+                "MyByteArrayConverter"
+            ).let { storeWithConverter ->
+                val intoStatement = storeWithConverter.findConverterIntoStatement(
+                    input = byteArray,
+                    columnTypes = listOf(
+                        string.makeNullable(),
+                        string.makeNonNullable()
+                    )
+                )
+                assertThat(intoStatement?.toSignature()).isEqualTo("fromByteArray")
+                assertThat(intoStatement?.to).isEqualTo(string.makeNonNullable())
+                assertThat(intoStatement?.from).isEqualTo(byteArray.makeNonNullable())
+                val fromCursor = storeWithConverter.findConverterFromCursor(
+                    output = byteArray,
+                    columnTypes = listOf(
+                        string.makeNullable(),
+                        string.makeNonNullable()
+                    )
+                )
+                assertThat(fromCursor?.toSignature()).isEqualTo("toByteArray")
+                assertThat(fromCursor?.to).isEqualTo(byteArray.makeNonNullable())
+                assertThat(fromCursor?.from).isEqualTo(string.makeNonNullable())
+            }
+        }
+    }
+
+    /**
+     * Collect results for conversion from String to our type
+     */
+    private fun collectStringConversionResults(
+        vararg selectedConverters: String
+    ): String {
+        val result = StringBuilder()
+        runProcessorTest(
+            sources = listOf(source),
+            options = mapOf(
+                USE_NULL_AWARE_CONVERTER.argName to "true"
+            )
+        ) { invocation ->
+            val store = invocation.createStore(*selectedConverters)
+            assertThat(store).isInstanceOf(NullAwareTypeConverterStore::class.java)
+            val myClassTypeElement = invocation.processingEnv.requireTypeElement(
+                "MyClass"
+            )
+            val stringTypeElement = invocation.processingEnv.requireTypeElement(
+                "java.lang.String"
+            )
+
+            result.appendLine(invocation.processingEnv.backend.name)
+            listOf(
+                stringTypeElement.type.makeNullable(),
+                stringTypeElement.type.makeNonNullable(),
+            ).forEach { stringType ->
+                listOf(
+                    myClassTypeElement.type.makeNullable(),
+                    myClassTypeElement.type.makeNonNullable()
+                ).forEach { myClassType ->
+                    val fromString = store.findTypeConverter(
+                        input = stringType,
+                        output = myClassType
+                    )
+                    val toString = store.findTypeConverter(
+                        input = myClassType,
+                        output = stringType
+                    )
+                    result.apply {
+                        append(stringType.toSignature())
+                        append(" to ")
+                        append(myClassType.toSignature())
+                        append(": ")
+                        appendLine(fromString?.toSignature() ?: "null")
+                    }
+
+                    result.apply {
+                        append(myClassType.toSignature())
+                        append(" to ")
+                        append(stringType.toSignature())
+                        append(": ")
+                        appendLine(toString?.toSignature() ?: "null")
+                    }
+                }
+            }
+        }
+        return result.toString()
+    }
+
+    /**
+     * Collect results for conversion from an unknown cursor type to our type
+     */
+    private fun collectCursorResults(
+        vararg selectedConverters: String
+    ): String {
+        val result = StringBuilder()
+        runProcessorTest(
+            sources = listOf(source),
+            options = mapOf(
+                USE_NULL_AWARE_CONVERTER.argName to "true"
+            )
+        ) { invocation ->
+            val store = invocation.createStore(*selectedConverters)
+            assertThat(store).isInstanceOf(NullAwareTypeConverterStore::class.java)
+            val myClassTypeElement = invocation.processingEnv.requireTypeElement(
+                "MyClass"
+            )
+
+            result.appendLine(invocation.processingEnv.backend.name)
+            listOf(
+                myClassTypeElement.type.makeNullable(),
+                myClassTypeElement.type.makeNonNullable()
+            ).forEach { myClassType ->
+                val toMyClass = store.findConverterFromCursor(
+                    columnTypes = null,
+                    output = myClassType
+                )
+                val fromMyClass = store.findConverterIntoStatement(
+                    input = myClassType,
+                    columnTypes = null
+                )
+
+                result.apply {
+                    append("Cursor to ")
+                    append(myClassType.toSignature())
+                    append(": ")
+                    appendLine(toMyClass?.toSignature() ?: "null")
+                }
+
+                result.apply {
+                    append(myClassType.toSignature())
+                    append(" to Cursor: ")
+                    appendLine(fromMyClass?.toSignature() ?: "null")
+                }
+            }
+        }
+        return result.toString()
+    }
+
+    private fun assertResult(result: String, expected: String) {
+        // remove commented lines from expected as they are used to explain cases for test's
+        // readability
+        assertThat(result).isEqualTo(
+            expected
+                .lines()
+                .filterNot { it.trim().startsWith("//") }
+                .joinToString("\n")
+        )
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/test/kotlin/androidx/room/solver/Signatures.kt b/room/room-compiler/src/test/kotlin/androidx/room/solver/Signatures.kt
new file mode 100644
index 0000000..4ddee69
--- /dev/null
+++ b/room/room-compiler/src/test/kotlin/androidx/room/solver/Signatures.kt
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver
+
+import androidx.room.compiler.processing.XNullability
+import androidx.room.compiler.processing.XType
+import androidx.room.solver.types.CompositeTypeConverter
+import androidx.room.solver.types.CustomTypeConverterWrapper
+import androidx.room.solver.types.NullSafeTypeConverter
+import androidx.room.solver.types.RequireNotNullTypeConverter
+import androidx.room.solver.types.TypeConverter
+import androidx.room.solver.types.UpCastTypeConverter
+
+// Shared signatures for objects that make testing more readable
+private fun XNullability.toSignature() = when (this) {
+    XNullability.NONNULL -> "!"
+    XNullability.NULLABLE -> "?"
+    XNullability.UNKNOWN -> ""
+}
+
+fun XType.toSignature() =
+    "$typeName${nullability.toSignature()}".substringAfter("java.lang.")
+
+fun TypeConverter.toSignature(): String {
+    return when (this) {
+        is CompositeTypeConverter -> "${conv1.toSignature()} / ${conv2.toSignature()}"
+        is CustomTypeConverterWrapper -> this.custom.methodName
+        is NullSafeTypeConverter ->
+            "(${this.from.toSignature()} == null " +
+                "? null : ${this.delegate.toSignature()})"
+        is RequireNotNullTypeConverter ->
+            "checkNotNull(${from.toSignature()})"
+        is UpCastTypeConverter ->
+            "(${from.toSignature()} as ${to.toSignature()})"
+        else -> "${from.toSignature()} -> ${to.toSignature()}"
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterCostTest.kt b/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterCostTest.kt
new file mode 100644
index 0000000..03ab939
--- /dev/null
+++ b/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterCostTest.kt
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2021 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room.solver
+
+import androidx.room.solver.types.TypeConverter.Cost
+import com.google.common.truth.Truth.assertThat
+import org.junit.Test
+
+class TypeConverterCostTest {
+    @Test
+    fun sections() {
+        val cost = Cost(
+            upCasts = 7,
+            nullSafeWrapper = 3,
+            converters = 1,
+            requireNotNull = 2
+        )
+        assertThat(
+            cost.toString()
+        ).isEqualTo(
+            "Cost[upcast:7,nullsafe:3,converters:1,requireNotNull:2]"
+        )
+    }
+
+    @Test
+    fun sum() {
+        assertThat(
+            Cost(converters = 3) + Cost(converters = 3, nullSafeWrapper = 4)
+
+        ).isEqualTo(Cost(converters = 6, nullSafeWrapper = 4))
+    }
+
+    @Test
+    fun sum2() {
+        val sum = Cost(
+            converters = 1,
+            upCasts = 60,
+        ) + Cost(
+            converters = 1,
+            nullSafeWrapper = 2,
+            upCasts = 20
+        )
+        assertThat(
+            sum
+        ).isEqualTo(
+            Cost(
+               converters = 2,
+               nullSafeWrapper = 2,
+               upCasts = 80
+            )
+        )
+    }
+
+    @Test
+    fun compare() {
+        assertThat(
+            Cost(requireNotNull = 1, converters = 2, nullSafeWrapper = 3, upCasts = 4)
+        ).isEquivalentAccordingToCompareTo(
+            Cost(requireNotNull = 1, converters = 2, nullSafeWrapper = 3, upCasts = 4)
+        )
+        // require not null is the most expensive
+        assertThat(
+            Cost(requireNotNull = 2, converters = 1, nullSafeWrapper = 1, upCasts = 1)
+        ).isGreaterThan(
+            Cost(requireNotNull = 1, converters = 2, nullSafeWrapper = 2, upCasts = 2)
+        )
+        // converters are the second most expensive
+        assertThat(
+            Cost(requireNotNull = 1, converters = 2, nullSafeWrapper = 1, upCasts = 1)
+        ).isGreaterThan(
+            Cost(requireNotNull = 1, converters = 1, nullSafeWrapper = 2, upCasts = 2)
+        )
+        // null safe wrapper is the third most expensive
+        assertThat(
+            Cost(requireNotNull = 1, converters = 1, nullSafeWrapper = 2, upCasts = 1)
+        ).isGreaterThan(
+            Cost(requireNotNull = 1, converters = 1, nullSafeWrapper = 1, upCasts = 2)
+        )
+        // upcast is the least expensive
+        assertThat(
+            Cost(requireNotNull = 1, converters = 1, nullSafeWrapper = 1, upCasts = 2)
+        ).isGreaterThan(
+            Cost(requireNotNull = 1, converters = 1, nullSafeWrapper = 1, upCasts = 1)
+        )
+    }
+}
\ No newline at end of file
diff --git a/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterStoreTest.kt b/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterStoreTest.kt
index 18c67d2..a6c6499 100644
--- a/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterStoreTest.kt
+++ b/room/room-compiler/src/test/kotlin/androidx/room/solver/TypeConverterStoreTest.kt
@@ -98,6 +98,14 @@
                 findConverter("Type1_Super", "Type2_Super")
             ).isEqualTo(
                 "Type1_Super -> JumpType_2 : JumpType_2 -> JumpType_3 : JumpType_3 -> Type2_Sub"
+                    .let { tillSub ->
+                        if (invocation.context.useNullAwareConverter) {
+                            // new type converter will have a step for upcasts too
+                            "$tillSub : Type2_Sub -> Type2_Super"
+                        } else {
+                            tillSub
+                        }
+                    }
             )
             assertThat(
                 findConverter("Type1", "Type2_Sub")
@@ -107,7 +115,14 @@
             assertThat(
                 findConverter("Type1_Sub", "Type2_Sub")
             ).isEqualTo(
-                "Type1 -> JumpType_1 : JumpType_1 -> Type2_Sub"
+                "Type1 -> JumpType_1 : JumpType_1 -> Type2_Sub".let { end ->
+                    if (invocation.context.useNullAwareConverter) {
+                        // new type converter will have a step for upcasts too
+                        "Type1_Sub -> Type1 : $end"
+                    } else {
+                        end
+                    }
+                }
             )
             assertThat(
                 findConverter("Type2", "Type2_Sub")