diff --git a/src/load.jl b/src/load.jl
index c2cb869..b5749ab 100644
--- a/src/load.jl
+++ b/src/load.jl
@@ -9,6 +9,7 @@ end
 
 sqltype(::Type{Union{T, Missing}}) where {T} = sqltype(T)
 sqltype(T) = get(SQLTYPES, T, "VARCHAR(255)")
+sqltype(T, coltypes, name) = get(coltypes, name, sqltype(T))
 
 const SQLTYPES = Dict{Type, String}(
     Int8 => "TINYINT",
@@ -34,26 +35,29 @@ const SQLTYPES = Dict{Type, String}(
 
 checkdupnames(names) = length(unique(map(x->lowercase(String(x)), names))) == length(names) || error("duplicate case-insensitive column names detected; sqlite doesn't allow duplicate column names and treats them case insensitive")
 
-function createtable(conn::Connection, nm::AbstractString, sch::Tables.Schema; debug::Bool=false, quoteidentifiers::Bool=true, createtableclause::AbstractString="CREATE TABLE", columnsuffix=Dict())
+function createtable(conn::Connection, nm::AbstractString, sch::Tables.Schema; debug::Bool=false, quoteidentifiers::Bool=true, createtableclause::AbstractString="CREATE TABLE", coltypes=Dict(), columnsuffix=Dict())
     names = sch.names
     checkdupnames(names)
-    types = [sqltype(T) for T in sch.types]
+    types = [sqltype(T, coltypes, names[i]) for (i, T) in enumerate(sch.types)]
     columns = (string(quoteidentifiers ? quoteid(String(names[i])) : names[i], ' ', types[i], ' ', get(columnsuffix, names[i], "")) for i = 1:length(names))
     debug && @info "executing create table statement: `$createtableclause $nm ($(join(columns, ", ")))`"
     return DBInterface.execute(conn, "$createtableclause $nm ($(join(columns, ", ")))")
 end
 
 """
-    MySQL.load(table, conn, name; append=true, quoteidentifiers=true, limit=typemax(Int64), createtableclause=nothing, columnsuffix=Dict(), debug=false)
-    table |> MySQL.load(conn, name; append=true, quoteidentifiers=true, limit=typemax(Int64), createtableclause=nothing, columnsuffix=Dict(), debug=false)
+    MySQL.load(table, conn, name; append=true, quoteidentifiers=true, limit=typemax(Int64), createtableclause=nothing, coltypes=Dict(), columnsuffix=Dict(), debug=false)
+    table |> MySQL.load(conn, name; append=true, quoteidentifiers=true, limit=typemax(Int64), createtableclause=nothing, coltypes=Dict(), columnsuffix=Dict(), debug=false)
 
 Attempts to take a Tables.jl source `table` and load into the database represented by `conn` with table name `name`.
 
 It first detects the `Tables.Schema` of the table source and generates a `CREATE TABLE` statement
-with the appropriate column names and types. If no table name is provided, one will be autogenerated, like `odbcjl_xxxxx`.
+with the appropriate column names and types. If no table name is provided, one will be autogenerated, like `mysql_xxxxx`.
 The `CREATE TABLE` clause can be provided manually by passing the `createtableclause` keyword argument, which
 would allow specifying a temporary table or `if not exists`.
-Column definitions can also be enhanced by providing arguments to `columnsuffix` as a `Dict` of 
+Column types can be overridden by providing the `coltypes` keyword argument as a `Dict` of
+column name (given as a `Symbol`) to a string of the SQL type. This allows, for example, using
+a `LONGBLOB` instead of `BLOB` for large binary data by doing `coltypes=Dict(:Photo => "LONGBLOB")`.
+Column definitions can also be enhanced by providing arguments to `columnsuffix` as a `Dict` of
 column name (given as a `Symbol`) to a string of the enhancement that will come after name and type like
 `[column name] [column type] enhancements`. This allows, for example, specifying the charset of a string column
 by doing something like `columnsuffix=Dict(:Name => "CHARACTER SET utf8mb4")`.
diff --git a/test/runtests.jl b/test/runtests.jl
index 344844e..436471b 100644
--- a/test/runtests.jl
+++ b/test/runtests.jl
@@ -161,6 +161,22 @@ ct0 = (x = UInt8[1, 22, 23],)
 MySQL.load(ct0,conn, debug=true)
 #there is no @test invocation here; but the above call failed due to Random not being imported
 
+# https://github.com/JuliaDatabases/MySQL.jl/issues/225
+# Test that coltypes parameter allows overriding default SQL types (e.g., BLOB -> LONGBLOB)
+ct225 = (
+    id = Int32[1, 2],
+    data = [UInt8[1, 2, 3], UInt8[4, 5, 6]],
+)
+MySQL.load(ct225, conn, "test225"; coltypes=Dict(:data => "LONGBLOB"), debug=true)
+col_info = DBInterface.execute(conn, "SHOW COLUMNS FROM test225 WHERE Field = 'data'") |> columntable
+# Type column may be returned as Vector{UInt8} or String depending on MySQL version
+col_type = col_info.Type[1]
+col_type_str = col_type isa Vector{UInt8} ? String(col_type) : col_type
+@test lowercase(col_type_str) == "longblob"
+# Also verify data roundtrips correctly
+ct225_roundtrip = DBInterface.execute(conn, "SELECT * FROM test225") |> columntable
+@test ct225_roundtrip.data == ct225.data
+
 # now test insert/parameter binding
 DBInterface.execute(conn, "DELETE FROM Employee")
 for i = 1:length(expected)