diff --git a/firebase-dataconnect/CHANGELOG.md b/firebase-dataconnect/CHANGELOG.md
index 1a8bb471488..6080722265a 100644
--- a/firebase-dataconnect/CHANGELOG.md
+++ b/firebase-dataconnect/CHANGELOG.md
@@ -6,6 +6,8 @@
   ([#6176](https://github.com/firebase/firebase-android-sdk/pull/6176))
 * [feature] Added `AnyValue` to support the `Any` custom GraphQL scalar type.
   ([#6285](https://github.com/firebase/firebase-android-sdk/pull/6285))
+* [feature] Added `OrderDirection` enum support.
+  ([#6307](https://github.com/firebase/firebase-android-sdk/pull/6307))
 * [feature] Added ability to specify `SerializersModule` when serializing.
   ([#6297](https://github.com/firebase/firebase-android-sdk/pull/6297))
 * [feature] Added `CallerSdkType`, which enables tracking of the generated SDK usage.
diff --git a/firebase-dataconnect/emulator/dataconnect/connector/alltypes/alltypes_ops.gql b/firebase-dataconnect/emulator/dataconnect/connector/alltypes/alltypes_ops.gql
index 575c2d64dfa..9ed5737e8c9 100644
--- a/firebase-dataconnect/emulator/dataconnect/connector/alltypes/alltypes_ops.gql
+++ b/firebase-dataconnect/emulator/dataconnect/connector/alltypes/alltypes_ops.gql
@@ -190,3 +190,33 @@ query getFarm($id: String!) @auth(level: PUBLIC) {
     }
   }
 }
+
+###############################################################################
+# Operations for table: OrderDirectionTest
+###############################################################################
+
+mutation OrderDirectionTestInsert5(
+  $tag: String!,
+  $value1: Int!,
+  $value2: Int!,
+  $value3: Int!,
+  $value4: Int!,
+  $value5: Int!,
+) @auth(level: PUBLIC) {
+  key1: orderDirectionTest_insert(data: { tag: $tag, value: $value1 })
+  key2: orderDirectionTest_insert(data: { tag: $tag, value: $value2 })
+  key3: orderDirectionTest_insert(data: { tag: $tag, value: $value3 })
+  key4: orderDirectionTest_insert(data: { tag: $tag, value: $value4 })
+  key5: orderDirectionTest_insert(data: { tag: $tag, value: $value5 })
+}
+
+query OrderDirectionTestGetAllByTag(
+  $tag: String!,
+  $orderDirection: OrderDirection,
+) @auth(level: PUBLIC) {
+  items: orderDirectionTests(
+    limit: 10,
+    orderBy: { value: $orderDirection },
+    where: { tag: { eq: $tag } },
+  ) { id }
+}
diff --git a/firebase-dataconnect/emulator/dataconnect/schema/alltypes_schema.gql b/firebase-dataconnect/emulator/dataconnect/schema/alltypes_schema.gql
index 47465c76f3a..279340d42d1 100644
--- a/firebase-dataconnect/emulator/dataconnect/schema/alltypes_schema.gql
+++ b/firebase-dataconnect/emulator/dataconnect/schema/alltypes_schema.gql
@@ -62,3 +62,8 @@ type Farmer @table {
   name: String!
   parent: Farmer
 }
+
+type OrderDirectionTest @table @index(fields: ["tag"]) {
+  value: Int
+  tag: String
+}
diff --git a/firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/OrderDirectionIntegrationTest.kt b/firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/OrderDirectionIntegrationTest.kt
new file mode 100644
index 00000000000..6dea7fd97af
--- /dev/null
+++ b/firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/OrderDirectionIntegrationTest.kt
@@ -0,0 +1,150 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * 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 com.google.firebase.dataconnect
+
+import com.google.firebase.dataconnect.testutil.DataConnectIntegrationTestBase
+import com.google.firebase.dataconnect.testutil.sortedParallelTo
+import com.google.firebase.dataconnect.testutil.tag
+import io.kotest.common.DelicateKotest
+import io.kotest.matchers.collections.shouldContainExactly
+import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder
+import io.kotest.property.Arb
+import io.kotest.property.arbitrary.distinct
+import io.kotest.property.arbitrary.int
+import io.kotest.property.arbitrary.next
+import kotlinx.coroutines.test.runTest
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.serializer
+import org.junit.Test
+
+class OrderDirectionIntegrationTest : DataConnectIntegrationTestBase() {
+
+  private val dataConnect: FirebaseDataConnect by lazy {
+    val connectorConfig = testConnectorConfig.copy(connector = "alltypes")
+    dataConnectFactory.newInstance(connectorConfig)
+  }
+
+  @OptIn(DelicateKotest::class) private val uniqueInts = Arb.int().distinct()
+
+  @Test
+  fun orderDirectionQueryVariableOmittedShouldUseAscendingOrder() = runTest {
+    val tag = Arb.tag().next(rs)
+    val values = List(5) { uniqueInts.next(rs) }
+    val insertedIds = insertRow(tag, values)
+
+    val queryIds = getRowIds(tag)
+
+    queryIds shouldContainExactlyInAnyOrder insertedIds
+  }
+
+  @Test
+  fun orderDirectionQueryVariableAscendingOrder() = runTest {
+    val tag = Arb.tag().next(rs)
+    val values = List(5) { uniqueInts.next(rs) }
+    val insertedIds = insertRow(tag, values)
+
+    val queryIds = getRowIds(tag, orderDirection = "ASC")
+
+    val insertedIdsSorted = insertedIds.sortedParallelTo(values)
+    queryIds shouldContainExactly insertedIdsSorted
+  }
+
+  @Test
+  fun orderDirectionQueryVariableDescendingOrder() = runTest {
+    val tag = Arb.tag().next(rs)
+    val values = List(5) { uniqueInts.next(rs) }
+    val insertedIds = insertRow(tag, values)
+
+    val queryIds = getRowIds(tag, orderDirection = "DESC")
+
+    val insertedIdsSorted = insertedIds.sortedParallelTo(values).reversed()
+    queryIds shouldContainExactly insertedIdsSorted
+  }
+
+  private suspend fun insertRow(tag: String, values: List<Int>): List<String> {
+    require(values.size == 5) { "values.size must be 5, but got ${values.size}" }
+    return insertRow(tag, values[0], values[1], values[2], values[3], values[4])
+  }
+
+  private suspend fun insertRow(
+    tag: String,
+    value1: Int,
+    value2: Int,
+    value3: Int,
+    value4: Int,
+    value5: Int
+  ): List<String> {
+    val variables = OrderDirectionTestInsert5Variables(tag, value1, value2, value3, value4, value5)
+    val mutationRef =
+      dataConnect.mutation<OrderDirectionTestInsert5Data, OrderDirectionTestInsert5Variables>(
+        operationName = "OrderDirectionTestInsert5",
+        variables = variables,
+        dataDeserializer = serializer(),
+        variablesSerializer = serializer(),
+      )
+    val result = mutationRef.execute()
+    return result.data.run { listOf(key1.id, key2.id, key3.id, key4.id, key5.id) }
+  }
+
+  private suspend fun getRowIds(tag: String, orderDirection: String? = null): List<String> {
+    val optionalOrderDirection =
+      if (orderDirection !== null) OptionalVariable.Value(orderDirection)
+      else OptionalVariable.Undefined
+    val variables = OrderDirectionTestGetAllByTagVariables(tag, optionalOrderDirection)
+    val queryRef =
+      dataConnect.query<OrderDirectionTestGetAllByTagData, OrderDirectionTestGetAllByTagVariables>(
+        operationName = "OrderDirectionTestGetAllByTag",
+        variables = variables,
+        dataDeserializer = serializer(),
+        variablesSerializer = serializer(),
+      )
+    val result = queryRef.execute()
+    return result.data.items.map { it.id }
+  }
+
+  @Serializable
+  data class OrderDirectionTestInsert5Variables(
+    val tag: String,
+    val value1: Int,
+    val value2: Int,
+    val value3: Int,
+    val value4: Int,
+    val value5: Int,
+  )
+
+  @Serializable data class OrderDirectionTestKey(val id: String)
+
+  @Serializable
+  data class OrderDirectionTestInsert5Data(
+    val key1: OrderDirectionTestKey,
+    val key2: OrderDirectionTestKey,
+    val key3: OrderDirectionTestKey,
+    val key4: OrderDirectionTestKey,
+    val key5: OrderDirectionTestKey,
+  )
+
+  @Serializable
+  data class OrderDirectionTestGetAllByTagVariables(
+    val tag: String,
+    val orderDirection: OptionalVariable<String>,
+  )
+
+  @Serializable
+  data class OrderDirectionTestGetAllByTagData(val items: List<Item>) {
+    @Serializable data class Item(val id: String)
+  }
+}
diff --git a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructDecoder.kt b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructDecoder.kt
index 1dbb0ea0ec5..10af8fc2b53 100644
--- a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructDecoder.kt
+++ b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructDecoder.kt
@@ -80,8 +80,8 @@ private object ProtoDecoderUtil {
   fun decodeDouble(value: Value, path: String?): Double =
     decode(value, path, KindCase.NUMBER_VALUE) { it.numberValue }
 
-  fun decodeEnum(value: Value, path: String?): Int =
-    decode(value, path, KindCase.NUMBER_VALUE) { it.numberValue.toInt() }
+  fun decodeEnum(value: Value, path: String?): String =
+    decode(value, path, KindCase.STRING_VALUE) { it.stringValue }
 
   fun decodeFloat(value: Value, path: String?): Float =
     decode(value, path, KindCase.NUMBER_VALUE) { it.numberValue.toFloat() }
@@ -134,7 +134,10 @@ internal class ProtoValueDecoder(
 
   override fun decodeDouble() = decodeDouble(valueProto, path)
 
-  override fun decodeEnum(enumDescriptor: SerialDescriptor) = decodeEnum(valueProto, path)
+  override fun decodeEnum(enumDescriptor: SerialDescriptor): Int {
+    val enumValueName = decodeEnum(valueProto, path)
+    return enumDescriptor.getElementIndex(enumValueName)
+  }
 
   override fun decodeFloat() = decodeFloat(valueProto, path)
 
diff --git a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructEncoder.kt b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructEncoder.kt
index 431f50159c0..5c442c95b9a 100644
--- a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructEncoder.kt
+++ b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/util/ProtoStructEncoder.kt
@@ -68,7 +68,7 @@ internal class ProtoValueEncoder(
   }
 
   override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) {
-    onValue(index.toValueProto())
+    onValue(enumDescriptor.getElementName(index).toValueProto())
   }
 
   override fun encodeFloat(value: Float) {
diff --git a/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/Arbs.kt b/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/Arbs.kt
index 836e3d99b2e..98a65552747 100644
--- a/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/Arbs.kt
+++ b/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/Arbs.kt
@@ -22,6 +22,7 @@ import com.google.firebase.dataconnect.FirebaseDataConnect.CallerSdkType
 import com.google.firebase.util.nextAlphanumericString
 import io.kotest.property.Arb
 import io.kotest.property.arbitrary.Codepoint
+import io.kotest.property.arbitrary.alphanumeric
 import io.kotest.property.arbitrary.arabic
 import io.kotest.property.arbitrary.arbitrary
 import io.kotest.property.arbitrary.ascii
@@ -163,3 +164,7 @@ fun <A> Arb<List<A>>.filterNotIncludesAllMatchingAnyScalars(values: List<Any?>)
 fun Arb.Companion.callerSdkType(): Arb<CallerSdkType> = arbitrary {
   if (Arb.boolean().bind()) CallerSdkType.Base else CallerSdkType.Generated
 }
+
+fun Arb.Companion.tag(): Arb<String> = arbitrary {
+  "tag" + Arb.string(size = 10, Codepoint.alphanumeric()).bind()
+}
diff --git a/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/TestUtils.kt b/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/TestUtils.kt
index 59825863067..c5e2f5fed07 100644
--- a/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/TestUtils.kt
+++ b/firebase-dataconnect/testutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/TestUtils.kt
@@ -231,3 +231,14 @@ fun TestScope.newBackgroundScopeThatAdvancesLikeForeground(): CoroutineScope {
     backgroundContextWithoutBackgroundWork + Job(backgroundContextWithoutBackgroundWork[Job])
   )
 }
+
+/** Sorts the given list and makes the same transformation on this list. */
+fun <T, U : Comparable<U>> List<T>.sortedParallelTo(other: List<U>): List<T> {
+  require(size == other.size) {
+    "size must equal other.size, but they are unequal: size=$size other.size=${other.size}"
+  }
+  val zippedList = other.zip(this)
+  val sortedZippedList = zippedList.sortedBy { it.first }
+  val (_, sortedThis) = sortedZippedList.unzip()
+  return sortedThis
+}