diff --git a/java/core/src/main/java/org/apache/arrow/adbc/core/TypedKey.java b/java/core/src/main/java/org/apache/arrow/adbc/core/TypedKey.java
index 1f1dda2f4b..957664ce48 100644
--- a/java/core/src/main/java/org/apache/arrow/adbc/core/TypedKey.java
+++ b/java/core/src/main/java/org/apache/arrow/adbc/core/TypedKey.java
@@ -42,6 +42,14 @@ public String getKey() {
     return key;
   }
 
+  public Class<T> getType() {
+    return type;
+  }
+
+  public T cast(Object value) {
+    return type.cast(value);
+  }
+
   /**
    * Get the option value (if it was set) and check the type.
    *
diff --git a/java/driver/jni/src/main/cpp/jni_wrapper.cc b/java/driver/jni/src/main/cpp/jni_wrapper.cc
index 080a310aec..7a1e2c6702 100644
--- a/java/driver/jni/src/main/cpp/jni_wrapper.cc
+++ b/java/driver/jni/src/main/cpp/jni_wrapper.cc
@@ -456,15 +456,163 @@ Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementExecuteSchema(
   return nullptr;
 }
 
+JNIEXPORT jbyteArray JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementGetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+
+  std::vector<uint8_t> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcStatementGetOptionBytes(db, key_str.value, const_cast<uint8_t*>(buf.data()),
+                                    &length, &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcStatementGetOptionBytes(db, key_str.value, const_cast<uint8_t*>(buf.data()),
+                                      &length, &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  jbyteArray result = env->NewByteArray(static_cast<jsize>(length));
+  env->SetByteArrayRegion(result, 0, static_cast<jsize>(length),
+                          reinterpret_cast<const jbyte*>(buf.data()));
+  return result;
+}
+
+JNIEXPORT jdouble JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementGetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+  double value = 0.0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcStatementGetOptionDouble(db, key_str.value, &value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0.0;
+  }
+  return static_cast<jdouble>(value);
+}
+
+JNIEXPORT jlong JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementGetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+  int64_t value = 0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcStatementGetOptionInt(db, key_str.value, &value, &error), error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0;
+  }
+  return static_cast<jlong>(value);
+}
+
+JNIEXPORT jstring JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementGetOptionString(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+
+  std::vector<char> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcStatementGetOption(db, key_str.value, const_cast<char*>(buf.data()), &length,
+                               &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcStatementGetOption(db, key_str.value, const_cast<char*>(buf.data()),
+                                 &length, &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  return env->NewStringUTF(buf.data());
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementSetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key,
+    jbyteArray value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    jsize value_length = env->GetArrayLength(value);
+    std::vector<uint8_t> value_buf(static_cast<size_t>(value_length));
+    env->GetByteArrayRegion(value, 0, value_length,
+                            reinterpret_cast<jbyte*>(value_buf.data()));
+    CHECK_ADBC_ERROR(AdbcStatementSetOptionBytes(db, key_str.value, value_buf.data(),
+                                                 value_buf.size(), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
 JNIEXPORT void JNICALL
-Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementSetOption(
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementSetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jdouble value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcStatementSetOptionDouble(db, key_str.value,
+                                                  static_cast<double>(value), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementSetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jlong value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcStatementSetOptionInt(db, key_str.value, static_cast<int64_t>(value), &error),
+        error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_statementSetOptionString(
     JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jstring value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
   try {
-    struct AdbcError error = ADBC_ERROR_INIT;
-    auto* ptr = reinterpret_cast<struct AdbcStatement*>(static_cast<uintptr_t>(handle));
     JniStringView key_str(env, key);
+    if (value == nullptr) {
+      CHECK_ADBC_ERROR(AdbcStatementSetOption(db, key_str.value, nullptr, &error), error);
+      return;
+    }
     JniStringView value_str(env, value);
-    CHECK_ADBC_ERROR(AdbcStatementSetOption(ptr, key_str.value, value_str.value, &error),
+    CHECK_ADBC_ERROR(AdbcStatementSetOption(db, key_str.value, value_str.value, &error),
                      error);
   } catch (const AdbcException& e) {
     e.ThrowJavaException(env);
@@ -598,4 +746,332 @@ Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionGetTableTypes(
   }
   return nullptr;
 }
+
+JNIEXPORT jbyteArray JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionGetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+
+  std::vector<uint8_t> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcConnectionGetOptionBytes(db, key_str.value, const_cast<uint8_t*>(buf.data()),
+                                     &length, &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcConnectionGetOptionBytes(db, key_str.value,
+                                       const_cast<uint8_t*>(buf.data()), &length, &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  jbyteArray result = env->NewByteArray(static_cast<jsize>(length));
+  env->SetByteArrayRegion(result, 0, static_cast<jsize>(length),
+                          reinterpret_cast<const jbyte*>(buf.data()));
+  return result;
+}
+
+JNIEXPORT jdouble JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionGetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  double value = 0.0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcConnectionGetOptionDouble(db, key_str.value, &value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0.0;
+  }
+  return static_cast<jdouble>(value);
+}
+
+JNIEXPORT jlong JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionGetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  int64_t value = 0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcConnectionGetOptionInt(db, key_str.value, &value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0;
+  }
+  return static_cast<jlong>(value);
+}
+
+JNIEXPORT jstring JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionGetOptionString(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+
+  std::vector<char> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcConnectionGetOption(db, key_str.value, const_cast<char*>(buf.data()), &length,
+                                &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcConnectionGetOption(db, key_str.value, const_cast<char*>(buf.data()),
+                                  &length, &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  return env->NewStringUTF(buf.data());
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionSetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key,
+    jbyteArray value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    jsize value_length = env->GetArrayLength(value);
+    std::vector<uint8_t> value_buf(static_cast<size_t>(value_length));
+    env->GetByteArrayRegion(value, 0, value_length,
+                            reinterpret_cast<jbyte*>(value_buf.data()));
+    CHECK_ADBC_ERROR(AdbcConnectionSetOptionBytes(db, key_str.value, value_buf.data(),
+                                                  value_buf.size(), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionSetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jdouble value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcConnectionSetOptionDouble(db, key_str.value,
+                                                   static_cast<double>(value), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionSetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jlong value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcConnectionSetOptionInt(db, key_str.value,
+                                                static_cast<int64_t>(value), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_connectionSetOptionString(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jstring value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcConnection*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    if (value == nullptr) {
+      CHECK_ADBC_ERROR(AdbcConnectionSetOption(db, key_str.value, nullptr, &error),
+                       error);
+      return;
+    }
+    JniStringView value_str(env, value);
+    CHECK_ADBC_ERROR(AdbcConnectionSetOption(db, key_str.value, value_str.value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT jbyteArray JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseGetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+
+  std::vector<uint8_t> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcDatabaseGetOptionBytes(db, key_str.value, const_cast<uint8_t*>(buf.data()),
+                                   &length, &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcDatabaseGetOptionBytes(db, key_str.value, const_cast<uint8_t*>(buf.data()),
+                                     &length, &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  jbyteArray result = env->NewByteArray(static_cast<jsize>(length));
+  env->SetByteArrayRegion(result, 0, static_cast<jsize>(length),
+                          reinterpret_cast<const jbyte*>(buf.data()));
+  return result;
+}
+
+JNIEXPORT jdouble JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseGetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  double value = 0.0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcDatabaseGetOptionDouble(db, key_str.value, &value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0.0;
+  }
+  return static_cast<jdouble>(value);
+}
+
+JNIEXPORT jlong JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseGetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  int64_t value = 0;
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcDatabaseGetOptionInt(db, key_str.value, &value, &error), error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return 0;
+  }
+  return static_cast<jlong>(value);
+}
+
+JNIEXPORT jstring JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseGetOptionString(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+
+  std::vector<char> buf(1024, '\0');
+  size_t length = buf.size();
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcDatabaseGetOption(db, key_str.value, const_cast<char*>(buf.data()), &length,
+                              &error),
+        error);
+    while (length > buf.size()) {
+      // Buffer was too small, resize and try again
+      buf.resize(length);
+      CHECK_ADBC_ERROR(
+          AdbcDatabaseGetOption(db, key_str.value, const_cast<char*>(buf.data()), &length,
+                                &error),
+          error);
+    }
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+    return nullptr;
+  }
+  return env->NewStringUTF(buf.data());
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseSetOptionBytes(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key,
+    jbyteArray value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    jsize value_length = env->GetArrayLength(value);
+    std::vector<uint8_t> value_buf(static_cast<size_t>(value_length));
+    env->GetByteArrayRegion(value, 0, value_length,
+                            reinterpret_cast<jbyte*>(value_buf.data()));
+    CHECK_ADBC_ERROR(AdbcDatabaseSetOptionBytes(db, key_str.value, value_buf.data(),
+                                                value_buf.size(), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseSetOptionDouble(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jdouble value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(AdbcDatabaseSetOptionDouble(db, key_str.value,
+                                                 static_cast<double>(value), &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseSetOptionLong(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jlong value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    CHECK_ADBC_ERROR(
+        AdbcDatabaseSetOptionInt(db, key_str.value, static_cast<int64_t>(value), &error),
+        error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
+
+JNIEXPORT void JNICALL
+Java_org_apache_arrow_adbc_driver_jni_impl_NativeAdbc_databaseSetOptionString(
+    JNIEnv* env, [[maybe_unused]] jclass self, jlong handle, jstring key, jstring value) {
+  struct AdbcError error = ADBC_ERROR_INIT;
+  auto* db = reinterpret_cast<struct AdbcDatabase*>(static_cast<uintptr_t>(handle));
+  try {
+    JniStringView key_str(env, key);
+    if (value == nullptr) {
+      CHECK_ADBC_ERROR(AdbcDatabaseSetOption(db, key_str.value, nullptr, &error), error);
+      return;
+    }
+    JniStringView value_str(env, value);
+    CHECK_ADBC_ERROR(AdbcDatabaseSetOption(db, key_str.value, value_str.value, &error),
+                     error);
+  } catch (const AdbcException& e) {
+    e.ThrowJavaException(env);
+  }
+}
 }
diff --git a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniConnection.java b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniConnection.java
index ef214eea32..008aeec427 100644
--- a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniConnection.java
+++ b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniConnection.java
@@ -21,6 +21,7 @@
 import org.apache.arrow.adbc.core.AdbcException;
 import org.apache.arrow.adbc.core.AdbcStatement;
 import org.apache.arrow.adbc.core.BulkIngestMode;
+import org.apache.arrow.adbc.core.TypedKey;
 import org.apache.arrow.adbc.driver.jni.impl.JniLoader;
 import org.apache.arrow.adbc.driver.jni.impl.NativeConnectionHandle;
 import org.apache.arrow.adbc.driver.jni.impl.NativeStatementHandle;
@@ -66,9 +67,9 @@ public AdbcStatement bulkIngest(String targetTableName, BulkIngestMode mode)
           throw new IllegalArgumentException("Unknown bulk ingest mode: " + mode);
       }
 
-      JniLoader.INSTANCE.statementSetOption(
+      JniLoader.INSTANCE.statementSetOptionString(
           stmtHandle, "adbc.ingest.target_table", targetTableName);
-      JniLoader.INSTANCE.statementSetOption(stmtHandle, "adbc.ingest.mode", modeValue);
+      JniLoader.INSTANCE.statementSetOptionString(stmtHandle, "adbc.ingest.mode", modeValue);
 
       return new JniStatement(allocator, stmtHandle);
     } catch (Exception e) {
@@ -120,4 +121,59 @@ public ArrowReader getTableTypes() throws AdbcException {
   public void close() {
     handle.close();
   }
+
+  @Override
+  public <T> T getOption(TypedKey<T> key) throws AdbcException {
+    if (key.getType() == String.class) {
+      return key.cast(JniLoader.INSTANCE.connectionGetOptionString(handle, key.getKey()));
+    } else if (key.getType() == Integer.class) {
+      return key.cast((int) JniLoader.INSTANCE.connectionGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Long.class) {
+      return key.cast(JniLoader.INSTANCE.connectionGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Float.class) {
+      return key.cast((float) JniLoader.INSTANCE.connectionGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Double.class) {
+      return key.cast(JniLoader.INSTANCE.connectionGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Boolean.class) {
+      String value = JniLoader.INSTANCE.connectionGetOptionString(handle, key.getKey());
+      if (value == null) {
+        return null;
+      } else if ("true".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.TRUE);
+      } else if ("false".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.FALSE);
+      } else {
+        throw AdbcException.invalidArgument(
+            "[jni] invalid boolean value for option " + key.getKey() + ": " + value);
+      }
+    } else if (key.getType() == byte[].class) {
+      return key.cast(JniLoader.INSTANCE.connectionGetOptionBytes(handle, key.getKey()));
+    }
+    return AdbcConnection.super.getOption(key);
+  }
+
+  @Override
+  public <T> void setOption(TypedKey<T> key, T value) throws AdbcException {
+    if (value instanceof String) {
+      JniLoader.INSTANCE.connectionSetOptionString(handle, key.getKey(), (String) value);
+    } else if (value == null) {
+      JniLoader.INSTANCE.connectionSetOptionString(handle, key.getKey(), null);
+    } else if (value instanceof Integer) {
+      JniLoader.INSTANCE.connectionSetOptionLong(handle, key.getKey(), (Integer) value);
+    } else if (value instanceof Long) {
+      JniLoader.INSTANCE.connectionSetOptionLong(handle, key.getKey(), (Long) value);
+    } else if (value instanceof Float) {
+      JniLoader.INSTANCE.connectionSetOptionDouble(handle, key.getKey(), (Float) value);
+    } else if (value instanceof Double) {
+      JniLoader.INSTANCE.connectionSetOptionDouble(handle, key.getKey(), (Double) value);
+    } else if (value instanceof Boolean) {
+      JniLoader.INSTANCE.connectionSetOptionString(
+          handle, key.getKey(), ((Boolean) value) ? "true" : "false");
+    } else if (value instanceof byte[]) {
+      JniLoader.INSTANCE.connectionSetOptionBytes(handle, key.getKey(), (byte[]) value);
+    } else {
+      throw AdbcException.invalidArgument(
+          "[jni] unsupported connection option type " + value.getClass());
+    }
+  }
 }
diff --git a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniDatabase.java b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniDatabase.java
index 1eaea0c5a0..9468107056 100644
--- a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniDatabase.java
+++ b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniDatabase.java
@@ -20,6 +20,7 @@
 import org.apache.arrow.adbc.core.AdbcConnection;
 import org.apache.arrow.adbc.core.AdbcDatabase;
 import org.apache.arrow.adbc.core.AdbcException;
+import org.apache.arrow.adbc.core.TypedKey;
 import org.apache.arrow.adbc.driver.jni.impl.JniLoader;
 import org.apache.arrow.adbc.driver.jni.impl.NativeDatabaseHandle;
 import org.apache.arrow.memory.BufferAllocator;
@@ -42,4 +43,59 @@ public AdbcConnection connect() throws AdbcException {
   public void close() {
     handle.close();
   }
+
+  @Override
+  public <T> T getOption(TypedKey<T> key) throws AdbcException {
+    if (key.getType() == String.class) {
+      return key.cast(JniLoader.INSTANCE.databaseGetOptionString(handle, key.getKey()));
+    } else if (key.getType() == Integer.class) {
+      return key.cast((int) JniLoader.INSTANCE.databaseGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Long.class) {
+      return key.cast(JniLoader.INSTANCE.databaseGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Float.class) {
+      return key.cast((float) JniLoader.INSTANCE.databaseGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Double.class) {
+      return key.cast(JniLoader.INSTANCE.databaseGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Boolean.class) {
+      String value = JniLoader.INSTANCE.databaseGetOptionString(handle, key.getKey());
+      if (value == null) {
+        return null;
+      } else if ("true".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.TRUE);
+      } else if ("false".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.FALSE);
+      } else {
+        throw AdbcException.invalidArgument(
+            "[jni] invalid boolean value for option " + key.getKey() + ": " + value);
+      }
+    } else if (key.getType() == byte[].class) {
+      return key.cast(JniLoader.INSTANCE.databaseGetOptionBytes(handle, key.getKey()));
+    }
+    return AdbcDatabase.super.getOption(key);
+  }
+
+  @Override
+  public <T> void setOption(TypedKey<T> key, T value) throws AdbcException {
+    if (value instanceof String) {
+      JniLoader.INSTANCE.databaseSetOptionString(handle, key.getKey(), (String) value);
+    } else if (value == null) {
+      JniLoader.INSTANCE.databaseSetOptionString(handle, key.getKey(), null);
+    } else if (value instanceof Integer) {
+      JniLoader.INSTANCE.databaseSetOptionLong(handle, key.getKey(), (Integer) value);
+    } else if (value instanceof Long) {
+      JniLoader.INSTANCE.databaseSetOptionLong(handle, key.getKey(), (Long) value);
+    } else if (value instanceof Float) {
+      JniLoader.INSTANCE.databaseSetOptionDouble(handle, key.getKey(), (Float) value);
+    } else if (value instanceof Double) {
+      JniLoader.INSTANCE.databaseSetOptionDouble(handle, key.getKey(), (Double) value);
+    } else if (value instanceof Boolean) {
+      JniLoader.INSTANCE.databaseSetOptionString(
+          handle, key.getKey(), ((Boolean) value) ? "true" : "false");
+    } else if (value instanceof byte[]) {
+      JniLoader.INSTANCE.databaseSetOptionBytes(handle, key.getKey(), (byte[]) value);
+    } else {
+      throw AdbcException.invalidArgument(
+          "[jni] unsupported database option type " + value.getClass());
+    }
+  }
 }
diff --git a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniStatement.java b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniStatement.java
index 2f98c3aa02..40a1eeec15 100644
--- a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniStatement.java
+++ b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/JniStatement.java
@@ -19,6 +19,7 @@
 
 import org.apache.arrow.adbc.core.AdbcException;
 import org.apache.arrow.adbc.core.AdbcStatement;
+import org.apache.arrow.adbc.core.TypedKey;
 import org.apache.arrow.adbc.driver.jni.impl.JniLoader;
 import org.apache.arrow.adbc.driver.jni.impl.NativeQueryResult;
 import org.apache.arrow.adbc.driver.jni.impl.NativeStatementHandle;
@@ -95,4 +96,59 @@ public void prepare() throws AdbcException {
   public void close() {
     handle.close();
   }
+
+  @Override
+  public <T> T getOption(TypedKey<T> key) throws AdbcException {
+    if (key.getType() == String.class) {
+      return key.cast(JniLoader.INSTANCE.statementGetOptionString(handle, key.getKey()));
+    } else if (key.getType() == Integer.class) {
+      return key.cast((int) JniLoader.INSTANCE.statementGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Long.class) {
+      return key.cast(JniLoader.INSTANCE.statementGetOptionLong(handle, key.getKey()));
+    } else if (key.getType() == Float.class) {
+      return key.cast((float) JniLoader.INSTANCE.statementGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Double.class) {
+      return key.cast(JniLoader.INSTANCE.statementGetOptionDouble(handle, key.getKey()));
+    } else if (key.getType() == Boolean.class) {
+      String value = JniLoader.INSTANCE.statementGetOptionString(handle, key.getKey());
+      if (value == null) {
+        return null;
+      } else if ("true".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.TRUE);
+      } else if ("false".equalsIgnoreCase(value)) {
+        return key.cast(Boolean.FALSE);
+      } else {
+        throw AdbcException.invalidArgument(
+            "[jni] invalid boolean value for option " + key.getKey() + ": " + value);
+      }
+    } else if (key.getType() == byte[].class) {
+      return key.cast(JniLoader.INSTANCE.statementGetOptionBytes(handle, key.getKey()));
+    }
+    return AdbcStatement.super.getOption(key);
+  }
+
+  @Override
+  public <T> void setOption(TypedKey<T> key, T value) throws AdbcException {
+    if (value instanceof String) {
+      JniLoader.INSTANCE.statementSetOptionString(handle, key.getKey(), (String) value);
+    } else if (value == null) {
+      JniLoader.INSTANCE.statementSetOptionString(handle, key.getKey(), null);
+    } else if (value instanceof Integer) {
+      JniLoader.INSTANCE.statementSetOptionLong(handle, key.getKey(), (Integer) value);
+    } else if (value instanceof Long) {
+      JniLoader.INSTANCE.statementSetOptionLong(handle, key.getKey(), (Long) value);
+    } else if (value instanceof Float) {
+      JniLoader.INSTANCE.statementSetOptionDouble(handle, key.getKey(), (Float) value);
+    } else if (value instanceof Double) {
+      JniLoader.INSTANCE.statementSetOptionDouble(handle, key.getKey(), (Double) value);
+    } else if (value instanceof Boolean) {
+      JniLoader.INSTANCE.statementSetOptionString(
+          handle, key.getKey(), ((Boolean) value) ? "true" : "false");
+    } else if (value instanceof byte[]) {
+      JniLoader.INSTANCE.statementSetOptionBytes(handle, key.getKey(), (byte[]) value);
+    } else {
+      throw AdbcException.invalidArgument(
+          "[jni] unsupported statement option type " + value.getClass());
+    }
+  }
 }
diff --git a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/JniLoader.java b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/JniLoader.java
index 8102af7a88..f1aa978d99 100644
--- a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/JniLoader.java
+++ b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/JniLoader.java
@@ -114,14 +114,49 @@ public void statementPrepare(NativeStatementHandle statement) throws AdbcExcepti
     NativeAdbc.statementPrepare(statement.getStatementHandle());
   }
 
-  public void statementSetOption(NativeStatementHandle statement, String key, String value)
+  public NativeSchemaResult statementExecuteSchema(NativeStatementHandle statement)
       throws AdbcException {
-    NativeAdbc.statementSetOption(statement.getStatementHandle(), key, value);
+    return NativeAdbc.statementExecuteSchema(statement.getStatementHandle());
   }
 
-  public NativeSchemaResult statementExecuteSchema(NativeStatementHandle statement)
+  public byte[] statementGetOptionBytes(NativeStatementHandle handle, String key)
       throws AdbcException {
-    return NativeAdbc.statementExecuteSchema(statement.getStatementHandle());
+    return NativeAdbc.statementGetOptionBytes(handle.getStatementHandle(), key);
+  }
+
+  public double statementGetOptionDouble(NativeStatementHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.statementGetOptionDouble(handle.getStatementHandle(), key);
+  }
+
+  public long statementGetOptionLong(NativeStatementHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.statementGetOptionLong(handle.getStatementHandle(), key);
+  }
+
+  public String statementGetOptionString(NativeStatementHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.statementGetOptionString(handle.getStatementHandle(), key);
+  }
+
+  public void statementSetOptionBytes(NativeStatementHandle handle, String key, byte[] value)
+      throws AdbcException {
+    NativeAdbc.statementSetOptionBytes(handle.getStatementHandle(), key, value);
+  }
+
+  public void statementSetOptionDouble(NativeStatementHandle handle, String key, double value)
+      throws AdbcException {
+    NativeAdbc.statementSetOptionDouble(handle.getStatementHandle(), key, value);
+  }
+
+  public void statementSetOptionLong(NativeStatementHandle handle, String key, long value)
+      throws AdbcException {
+    NativeAdbc.statementSetOptionLong(handle.getStatementHandle(), key, value);
+  }
+
+  public void statementSetOptionString(NativeStatementHandle statement, String key, String value)
+      throws AdbcException {
+    NativeAdbc.statementSetOptionString(statement.getStatementHandle(), key, value);
   }
 
   public NativeQueryResult connectionGetObjects(
@@ -159,4 +194,83 @@ public NativeQueryResult connectionGetTableTypes(NativeConnectionHandle connecti
       throws AdbcException {
     return NativeAdbc.connectionGetTableTypes(connection.getConnectionHandle());
   }
+
+  public byte[] connectionGetOptionBytes(NativeConnectionHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.connectionGetOptionBytes(handle.getConnectionHandle(), key);
+  }
+
+  public double connectionGetOptionDouble(NativeConnectionHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.connectionGetOptionDouble(handle.getConnectionHandle(), key);
+  }
+
+  public long connectionGetOptionLong(NativeConnectionHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.connectionGetOptionLong(handle.getConnectionHandle(), key);
+  }
+
+  public String connectionGetOptionString(NativeConnectionHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.connectionGetOptionString(handle.getConnectionHandle(), key);
+  }
+
+  public void connectionSetOptionBytes(NativeConnectionHandle handle, String key, byte[] value)
+      throws AdbcException {
+    NativeAdbc.connectionSetOptionBytes(handle.getConnectionHandle(), key, value);
+  }
+
+  public void connectionSetOptionDouble(NativeConnectionHandle handle, String key, double value)
+      throws AdbcException {
+    NativeAdbc.connectionSetOptionDouble(handle.getConnectionHandle(), key, value);
+  }
+
+  public void connectionSetOptionLong(NativeConnectionHandle handle, String key, long value)
+      throws AdbcException {
+    NativeAdbc.connectionSetOptionLong(handle.getConnectionHandle(), key, value);
+  }
+
+  public void connectionSetOptionString(NativeConnectionHandle connection, String key, String value)
+      throws AdbcException {
+    NativeAdbc.connectionSetOptionString(connection.getConnectionHandle(), key, value);
+  }
+
+  public byte[] databaseGetOptionBytes(NativeDatabaseHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.databaseGetOptionBytes(handle.getDatabaseHandle(), key);
+  }
+
+  public double databaseGetOptionDouble(NativeDatabaseHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.databaseGetOptionDouble(handle.getDatabaseHandle(), key);
+  }
+
+  public long databaseGetOptionLong(NativeDatabaseHandle handle, String key) throws AdbcException {
+    return NativeAdbc.databaseGetOptionLong(handle.getDatabaseHandle(), key);
+  }
+
+  public String databaseGetOptionString(NativeDatabaseHandle handle, String key)
+      throws AdbcException {
+    return NativeAdbc.databaseGetOptionString(handle.getDatabaseHandle(), key);
+  }
+
+  public void databaseSetOptionBytes(NativeDatabaseHandle handle, String key, byte[] value)
+      throws AdbcException {
+    NativeAdbc.databaseSetOptionBytes(handle.getDatabaseHandle(), key, value);
+  }
+
+  public void databaseSetOptionDouble(NativeDatabaseHandle handle, String key, double value)
+      throws AdbcException {
+    NativeAdbc.databaseSetOptionDouble(handle.getDatabaseHandle(), key, value);
+  }
+
+  public void databaseSetOptionLong(NativeDatabaseHandle handle, String key, long value)
+      throws AdbcException {
+    NativeAdbc.databaseSetOptionLong(handle.getDatabaseHandle(), key, value);
+  }
+
+  public void databaseSetOptionString(NativeDatabaseHandle handle, String key, String value)
+      throws AdbcException {
+    NativeAdbc.databaseSetOptionString(handle.getDatabaseHandle(), key, value);
+  }
 }
diff --git a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/NativeAdbc.java b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/NativeAdbc.java
index 2dcf70f625..3a6d09c8ce 100644
--- a/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/NativeAdbc.java
+++ b/java/driver/jni/src/main/java/org/apache/arrow/adbc/driver/jni/impl/NativeAdbc.java
@@ -50,10 +50,28 @@ static native NativeDatabaseHandle openDatabase(int version, String[] parameters
 
   static native void statementPrepare(long handle) throws AdbcException;
 
-  static native void statementSetOption(long handle, String key, String value) throws AdbcException;
-
   static native NativeSchemaResult statementExecuteSchema(long handle) throws AdbcException;
 
+  static native byte[] statementGetOptionBytes(long handle, String key) throws AdbcException;
+
+  static native double statementGetOptionDouble(long handle, String key) throws AdbcException;
+
+  static native long statementGetOptionLong(long handle, String key) throws AdbcException;
+
+  static native String statementGetOptionString(long handle, String key) throws AdbcException;
+
+  static native void statementSetOptionBytes(long handle, String key, byte[] value)
+      throws AdbcException;
+
+  static native void statementSetOptionDouble(long handle, String key, double value)
+      throws AdbcException;
+
+  static native void statementSetOptionLong(long handle, String key, long value)
+      throws AdbcException;
+
+  static native void statementSetOptionString(long handle, String key, String value)
+      throws AdbcException;
+
   static native NativeQueryResult connectionGetObjects(
       long handle,
       int depth,
@@ -71,4 +89,44 @@ static native NativeSchemaResult connectionGetTableSchema(
       long handle, String catalog, String dbSchema, String tableName) throws AdbcException;
 
   static native NativeQueryResult connectionGetTableTypes(long handle) throws AdbcException;
+
+  static native byte[] connectionGetOptionBytes(long handle, String key) throws AdbcException;
+
+  static native double connectionGetOptionDouble(long handle, String key) throws AdbcException;
+
+  static native long connectionGetOptionLong(long handle, String key) throws AdbcException;
+
+  static native String connectionGetOptionString(long handle, String key) throws AdbcException;
+
+  static native void connectionSetOptionBytes(long handle, String key, byte[] value)
+      throws AdbcException;
+
+  static native void connectionSetOptionDouble(long handle, String key, double value)
+      throws AdbcException;
+
+  static native void connectionSetOptionLong(long handle, String key, long value)
+      throws AdbcException;
+
+  static native void connectionSetOptionString(long handle, String key, String value)
+      throws AdbcException;
+
+  static native byte[] databaseGetOptionBytes(long handle, String key) throws AdbcException;
+
+  static native double databaseGetOptionDouble(long handle, String key) throws AdbcException;
+
+  static native long databaseGetOptionLong(long handle, String key) throws AdbcException;
+
+  static native String databaseGetOptionString(long handle, String key) throws AdbcException;
+
+  static native void databaseSetOptionBytes(long handle, String key, byte[] value)
+      throws AdbcException;
+
+  static native void databaseSetOptionDouble(long handle, String key, double value)
+      throws AdbcException;
+
+  static native void databaseSetOptionLong(long handle, String key, long value)
+      throws AdbcException;
+
+  static native void databaseSetOptionString(long handle, String key, String value)
+      throws AdbcException;
 }
diff --git a/java/driver/jni/src/test/java/org/apache/arrow/adbc/driver/jni/JniDriverTest.java b/java/driver/jni/src/test/java/org/apache/arrow/adbc/driver/jni/JniDriverTest.java
index c61fa8d2b9..c9a0922210 100644
--- a/java/driver/jni/src/test/java/org/apache/arrow/adbc/driver/jni/JniDriverTest.java
+++ b/java/driver/jni/src/test/java/org/apache/arrow/adbc/driver/jni/JniDriverTest.java
@@ -25,12 +25,14 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.LongStream;
+import java.util.stream.Stream;
 import org.apache.arrow.adbc.core.AdbcConnection;
 import org.apache.arrow.adbc.core.AdbcDatabase;
 import org.apache.arrow.adbc.core.AdbcDriver;
@@ -50,6 +52,8 @@
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 class JniDriverTest {
   @Test
@@ -91,7 +95,7 @@ void loadProfile(@TempDir Path tempDir) throws Exception {
                 "profile_version = 1",
                 "driver = \"adbc_driver_sqlite\"",
                 "[Options]",
-                "adbc.sqlite.query.batch_rows = 1024")
+                "adbc.sqlite.query.batch_rows = 4242")
             .getBytes(StandardCharsets.UTF_8));
     try (final BufferAllocator allocator = new RootAllocator()) {
       JniDriver driver = new JniDriver(allocator);
@@ -99,12 +103,8 @@ void loadProfile(@TempDir Path tempDir) throws Exception {
       JniDriver.PARAM_PROFILE.set(parameters, "myprofile");
       JniDriver.PARAM_PROFILE_SEARCH_PATH.set(parameters, tempDir.toString());
       try (final AdbcDatabase db = driver.open(parameters)) {
-        // TODO(lidavidm): getOption not implemented
-        AdbcException e =
-            assertThrows(
-                AdbcException.class,
-                () -> db.getOption(new TypedKey<>("adbc.sqlite.query.batch_rows", Long.class)));
-        assertThat(e).hasMessageContaining("Unsupported option");
+        assertThat(db.getOption(new TypedKey<>("adbc.sqlite.query.batch_rows", Long.class)))
+            .isEqualTo(4242L);
       }
     }
   }
@@ -398,4 +398,135 @@ void bulkIngest() throws Exception {
       }
     }
   }
+
+  @Test
+  void getSetOption() throws Exception {
+    TypedKey<Integer> batchRowsInt = new TypedKey<>("adbc.sqlite.query.batch_rows", Integer.class);
+    TypedKey<Long> batchRowsLong = new TypedKey<>("adbc.sqlite.query.batch_rows", Long.class);
+    TypedKey<Boolean> bindByName = new TypedKey<>("adbc.statement.bind_by_name", Boolean.class);
+    TypedKey<String> bindByNameString = new TypedKey<>("adbc.statement.bind_by_name", String.class);
+    try (final BufferAllocator allocator = new RootAllocator()) {
+      JniDriver driver = new JniDriver(allocator);
+      Map<String, Object> parameters = new HashMap<>();
+      JniDriver.PARAM_DRIVER.set(parameters, "adbc_driver_sqlite");
+      try (final AdbcDatabase db = driver.open(parameters)) {
+        assertThat(db.getOption(batchRowsInt)).isEqualTo(1024);
+        assertThat(db.getOption(batchRowsLong)).isEqualTo(1024L);
+        assertThat(db.getOption(AdbcDriver.PARAM_URI))
+            .isEqualTo("file:adbc_driver_sqlite?mode=memory&cache=shared");
+
+        try (final AdbcConnection conn = db.connect();
+            final AdbcStatement stmt = conn.createStatement()) {
+          assertThat(conn.getOption(batchRowsInt)).isEqualTo(1024);
+          assertThat(conn.getOption(batchRowsLong)).isEqualTo(1024L);
+
+          assertThat(stmt.getOption(batchRowsInt)).isEqualTo(1024);
+          assertThat(stmt.getOption(batchRowsLong)).isEqualTo(1024L);
+          stmt.setOption(batchRowsLong, 42L);
+          assertThat(stmt.getOption(batchRowsLong)).isEqualTo(42L);
+          assertThat(stmt.getOption(bindByName)).isFalse();
+          assertThat(stmt.getOption(bindByNameString)).isEqualTo("false");
+          stmt.setOption(bindByName, true);
+          assertThat(stmt.getOption(bindByName)).isTrue();
+          assertThat(stmt.getOption(bindByNameString)).isEqualTo("true");
+        }
+      }
+    }
+  }
+
+  static class GetSetOptionFailCase {
+    @SuppressWarnings("rawtypes")
+    final TypedKey key;
+
+    final Object value;
+    final String message;
+
+    GetSetOptionFailCase(TypedKey<?> key, Object value, String message) {
+      this.key = key;
+      this.value = value;
+      this.message = message;
+    }
+
+    @Override
+    public String toString() {
+      String v;
+      if (value == null) {
+        v = "(NULL)";
+      } else if (value instanceof byte[]) {
+        v = Arrays.toString((byte[]) value);
+      } else {
+        v = value.toString();
+      }
+      return "key=" + key.getKey() + ", value=" + v;
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("getSetOptionFailProvider")
+  void getSetOptionFailDatabase(GetSetOptionFailCase testCase) throws Exception {
+    // These will fail; we don't have a driver that supports an example of every type
+    try (final BufferAllocator allocator = new RootAllocator()) {
+      JniDriver driver = new JniDriver(allocator);
+      Map<String, Object> parameters = new HashMap<>();
+      JniDriver.PARAM_DRIVER.set(parameters, "adbc_driver_sqlite");
+      try (final AdbcDatabase db = driver.open(parameters)) {
+        AdbcException e;
+        //noinspection unchecked
+        e = assertThrows(AdbcException.class, () -> db.setOption(testCase.key, testCase.value));
+        assertThat(e).hasMessageContaining(testCase.message);
+      }
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("getSetOptionFailProvider")
+  void getSetOptionFailConnection(GetSetOptionFailCase testCase) throws Exception {
+    try (final BufferAllocator allocator = new RootAllocator()) {
+      JniDriver driver = new JniDriver(allocator);
+      Map<String, Object> parameters = new HashMap<>();
+      JniDriver.PARAM_DRIVER.set(parameters, "adbc_driver_sqlite");
+      try (final AdbcDatabase db = driver.open(parameters);
+          final AdbcConnection conn = db.connect()) {
+        AdbcException e;
+        //noinspection unchecked
+        e = assertThrows(AdbcException.class, () -> conn.setOption(testCase.key, testCase.value));
+        assertThat(e).hasMessageContaining(testCase.message);
+      }
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("getSetOptionFailProvider")
+  void getSetOptionFailStatement(GetSetOptionFailCase testCase) throws Exception {
+    try (final BufferAllocator allocator = new RootAllocator()) {
+      JniDriver driver = new JniDriver(allocator);
+      Map<String, Object> parameters = new HashMap<>();
+      JniDriver.PARAM_DRIVER.set(parameters, "adbc_driver_sqlite");
+      try (final AdbcDatabase db = driver.open(parameters);
+          final AdbcConnection conn = db.connect();
+          final AdbcStatement stmt = conn.createStatement()) {
+        AdbcException e;
+        //noinspection unchecked
+        e = assertThrows(AdbcException.class, () -> stmt.setOption(testCase.key, testCase.value));
+        assertThat(e).hasMessageContaining(testCase.message);
+      }
+    }
+  }
+
+  static Stream<GetSetOptionFailCase> getSetOptionFailProvider() {
+    return Stream.of(
+        new GetSetOptionFailCase(new TypedKey<>("unknown", Integer.class), 2048, "unknown=2048"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", Long.class), 2048L, "unknown=2048"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", Float.class), 2048f, "unknown=2048.0"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", Double.class), 2048d, "unknown=2048.0"),
+        new GetSetOptionFailCase(
+            new TypedKey<>("unknown", String.class), "foobar", "unknown='foobar'"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", String.class), null, "unknown=(NULL)"),
+        new GetSetOptionFailCase(
+            new TypedKey<>("unknown", Boolean.class), false, "unknown='false'"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", Boolean.class), true, "unknown='true'"),
+        new GetSetOptionFailCase(
+            new TypedKey<>("unknown", byte[].class), new byte[] {0, 42, 0}, "unknown=(3 bytes)"),
+        new GetSetOptionFailCase(new TypedKey<>("unknown", byte[].class), null, "unknown=(NULL)"));
+  }
 }