diff --git a/dashboard/components/BackPressureTable.tsx b/dashboard/components/BackPressureTable.tsx
index 9629a429d9f95..241fde0ddc13c 100644
--- a/dashboard/components/BackPressureTable.tsx
+++ b/dashboard/components/BackPressureTable.tsx
@@ -24,11 +24,11 @@ import {
   Th,
   Thead,
   Tr,
-  useToast,
 } from "@chakra-ui/react"
 import { sortBy } from "lodash"
 import Head from "next/head"
 import { Fragment, useEffect, useState } from "react"
+import useErrorToast from "../hook/useErrorToast"
 import { getActorBackPressures } from "../pages/api/metric"
 import RateBar from "./RateBar"
 import { Metrics } from "./metrics"
@@ -44,7 +44,7 @@ export default function BackPressureTable({
 }) {
   const [backPressuresMetrics, setBackPressuresMetrics] =
     useState<BackPressuresMetrics>()
-  const toast = useToast()
+  const toast = useErrorToast()
 
   useEffect(() => {
     async function doFetch() {
@@ -58,14 +58,7 @@ export default function BackPressureTable({
           setBackPressuresMetrics(metrics)
           await new Promise((resolve) => setTimeout(resolve, 5000)) // refresh every 5 secs
         } catch (e: any) {
-          toast({
-            title: "Error Occurred",
-            description: e.toString(),
-            status: "error",
-            duration: 5000,
-            isClosable: true,
-          })
-          console.error(e)
+          toast(e, "warning")
           break
         }
       }
diff --git a/dashboard/components/FragmentGraph.tsx b/dashboard/components/FragmentGraph.tsx
index e188310910638..891a6d21df4b0 100644
--- a/dashboard/components/FragmentGraph.tsx
+++ b/dashboard/components/FragmentGraph.tsx
@@ -20,7 +20,7 @@ import {
   generateBoxLinks,
   layout,
 } from "../lib/layout"
-import { PlanNodeDatum } from "../pages/streaming_plan"
+import { PlanNodeDatum } from "../pages/fragment_graph"
 import BackPressureTable from "./BackPressureTable"
 
 const ReactJson = loadable(() => import("react-json-view"))
@@ -218,7 +218,7 @@ export default function FragmentGraph({
 
         text
           .attr("fill", "black")
-          .text(({ id }) => `Fragment #${id}`)
+          .text(({ id }) => `Fragment ${id}`)
           .attr("font-family", "inherit")
           .attr("text-anchor", "end")
           .attr("dy", ({ height }) => height - actorMarginY + 12)
diff --git a/dashboard/components/Layout.tsx b/dashboard/components/Layout.tsx
index c161eefa20a33..a0bc7acd730d4 100644
--- a/dashboard/components/Layout.tsx
+++ b/dashboard/components/Layout.tsx
@@ -31,6 +31,7 @@ import { UrlObject } from "url"
 import {
   IconArrowRightCircle,
   IconArrowRightCircleFill,
+  IconBoxArrowUpRight,
   IconServer,
 } from "../components/utils/icons"
 
@@ -41,11 +42,13 @@ function NavButton({
   children,
   leftIcon,
   leftIconActive,
+  external,
 }: {
   href: string | UrlObject
   children?: React.ReactNode
   leftIcon?: React.ReactElement
   leftIconActive?: React.ReactElement
+  external?: boolean
 }) {
   const router = useRouter()
   const [match, setMatch] = useState(false)
@@ -55,19 +58,22 @@ function NavButton({
     return () => {}
   }, [href, router.asPath])
 
+  const icon =
+    leftIcon || (external ? <IconBoxArrowUpRight /> : <IconArrowRightCircle />)
+  const activeIcon =
+    leftIconActive ||
+    leftIcon ||
+    (external ? undefined : <IconArrowRightCircleFill />)
+
   return (
-    <Link href={href}>
+    <Link href={href} target={external ? "_blank" : undefined}>
       <Button
         colorScheme={match ? "blue" : "gray"}
         color={match ? "blue.600" : "gray.500"}
         variant={match ? "outline" : "ghost"}
         width="full"
         justifyContent="flex-start"
-        leftIcon={
-          match
-            ? leftIconActive || leftIcon || <IconArrowRightCircleFill />
-            : leftIcon || <IconArrowRightCircle />
-        }
+        leftIcon={match ? activeIcon : icon}
       >
         {children}
       </Button>
@@ -83,6 +89,14 @@ function NavTitle({ children }: { children: React.ReactNode }) {
   )
 }
 
+function Section({ children }: { children: React.ReactNode }) {
+  return (
+    <VStack width="full" alignItems="flex-start" px={3}>
+      {children}
+    </VStack>
+  )
+}
+
 function Layout({ children }: { children: React.ReactNode }) {
   return (
     <Flex>
@@ -110,10 +124,12 @@ function Layout({ children }: { children: React.ReactNode }) {
               </Text>
             </HStack>
           </Box>
-          <NavButton href="/cluster/" leftIcon={<IconServer />}>
-            Cluster Overview
-          </NavButton>
-          <VStack width="full" alignItems="flex-start" px={3}>
+          <Section>
+            <NavButton href="/cluster/" leftIcon={<IconServer />}>
+              Cluster Overview
+            </NavButton>
+          </Section>
+          <Section>
             <NavTitle>Catalog</NavTitle>
             <NavButton href="/sources/">Sources</NavButton>
             <NavButton href="/tables/">Tables</NavButton>
@@ -124,28 +140,32 @@ function Layout({ children }: { children: React.ReactNode }) {
             <NavButton href="/internal_tables/">Internal Tables</NavButton>
             <NavButton href="/sinks/">Sinks</NavButton>
             <NavButton href="/views/">Views</NavButton>
-          </VStack>
-          <VStack width="full" alignItems="flex-start" px={3}>
+          </Section>
+          <Section>
             <NavTitle>Streaming</NavTitle>
-            <NavButton href="/streaming_graph/">Graph</NavButton>
-            <NavButton href="/streaming_plan/">Fragments</NavButton>
-          </VStack>
-          <VStack width="full" alignItems="flex-start" px={3}>
+            <NavButton href="/dependency_graph/">Dependency Graph</NavButton>
+            <NavButton href="/fragment_graph/">Fragment Graph</NavButton>
+          </Section>
+          <Section>
             <NavTitle>Batch</NavTitle>
             <NavButton href="/batch_tasks/">Batch Tasks</NavButton>
-          </VStack>
-          <VStack width="full" alignItems="flex-start" px={3}>
+          </Section>
+          <Section>
             <NavTitle>Explain</NavTitle>
             <NavButton href="/explain_distsql/">Distributed Plan</NavButton>
-          </VStack>
-          <VStack width="full" alignItems="flex-start" px={3}>
+          </Section>
+          <Section>
             <NavTitle>Debug</NavTitle>
             <NavButton href="/await_tree/">Await Tree Dump</NavButton>
             <NavButton href="/heap_profiling/">Heap Profiling</NavButton>
-            <NavButton href="/api/monitor/diagnose">Diagnose</NavButton>
-          </VStack>
-          <VStack mb={3}></VStack>
-          <NavButton href="/settings/">Settings</NavButton>
+            <NavButton href="/api/monitor/diagnose" external>
+              Diagnose
+            </NavButton>
+          </Section>
+          <Section>
+            <NavTitle>Settings</NavTitle>
+            <NavButton href="/settings/">Settings</NavButton>
+          </Section>
         </VStack>
       </Box>
       <Box flex={1} overflowY="scroll">
diff --git a/dashboard/components/Relations.tsx b/dashboard/components/Relations.tsx
index 33df34b753b7a..acdba0e2608c2 100644
--- a/dashboard/components/Relations.tsx
+++ b/dashboard/components/Relations.tsx
@@ -33,7 +33,6 @@ import {
   Thead,
   Tr,
   useDisclosure,
-  useToast,
 } from "@chakra-ui/react"
 import loadable from "@loadable/component"
 import Head from "next/head"
@@ -41,6 +40,7 @@ import Head from "next/head"
 import Link from "next/link"
 import { Fragment, useEffect, useState } from "react"
 import Title from "../components/Title"
+import useErrorToast from "../hook/useErrorToast"
 import extractColumnInfo from "../lib/extractInfo"
 import { Relation, StreamingJob } from "../pages/api/streaming"
 import {
@@ -121,7 +121,7 @@ export function Relations<R extends Relation>(
   getRelations: () => Promise<R[]>,
   extraColumns: Column<R>[]
 ) {
-  const toast = useToast()
+  const toast = useErrorToast()
   const [relationList, setRelationList] = useState<R[]>([])
 
   useEffect(() => {
@@ -129,14 +129,7 @@ export function Relations<R extends Relation>(
       try {
         setRelationList(await getRelations())
       } catch (e: any) {
-        toast({
-          title: "Error Occurred",
-          description: e.toString(),
-          status: "error",
-          duration: 5000,
-          isClosable: true,
-        })
-        console.error(e)
+        toast(e)
       }
     }
     doFetch()
diff --git a/dashboard/components/utils/icons.tsx b/dashboard/components/utils/icons.tsx
index a3e32af20fbd4..8bef660526c95 100644
--- a/dashboard/components/utils/icons.tsx
+++ b/dashboard/components/utils/icons.tsx
@@ -26,3 +26,7 @@ export function IconArrowRightCircle() {
 export function IconArrowRightCircleFill() {
   return <i className="bi bi-arrow-right-circle-fill"></i>
 }
+
+export function IconBoxArrowUpRight() {
+  return <i className="bi bi-box-arrow-up-right"></i>
+}
diff --git a/dashboard/hook/useErrorToast.ts b/dashboard/hook/useErrorToast.ts
new file mode 100644
index 0000000000000..1c86cc3084fbf
--- /dev/null
+++ b/dashboard/hook/useErrorToast.ts
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2024 RisingWave Labs
+ *
+ * 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.
+ */
+
+import { AlertStatus, useToast } from "@chakra-ui/react"
+import { useCallback } from "react"
+
+export default function useErrorToast() {
+  const toast = useToast()
+
+  return useCallback(
+    (e: any, status: AlertStatus = "error") => {
+      let title: string
+      let description: string | undefined
+
+      if (e instanceof Error) {
+        title = e.message
+        description = e.cause?.toString()
+      } else {
+        title = e.toString()
+      }
+
+      toast({
+        title,
+        description,
+        status,
+        duration: 5000,
+        isClosable: true,
+      })
+
+      console.error(e)
+    },
+    [toast]
+  )
+}
diff --git a/dashboard/hook/useWindowSize.js b/dashboard/hook/useWindowSize.js
deleted file mode 100644
index 2b5a35d0cd710..0000000000000
--- a/dashboard/hook/useWindowSize.js
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * 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.
- *
- */
-import { useEffect, useState } from "react"
-
-export default function useWindowSize() {
-  const [size, setSize] = useState({ w: 1024, h: 768 })
-  // setSize({
-  //   w: window.innerWidth,
-  //   h: window.innerHeight
-  // })
-  const setSizeOnEvent = () => {
-    setSize({
-      w: window.innerWidth,
-      h: window.innerHeight,
-    })
-  }
-  useEffect(() => {
-    window.addEventListener("resize", setSizeOnEvent)
-    return () => window.removeEventListener("resize", setSizeOnEvent)
-  }, [])
-  return size
-}
diff --git a/dashboard/pages/_app.tsx b/dashboard/pages/_app.tsx
index e383c2d71fae4..35f46f222cd7c 100644
--- a/dashboard/pages/_app.tsx
+++ b/dashboard/pages/_app.tsx
@@ -30,7 +30,11 @@ function App({ Component, pageProps }: AppProps) {
   const [isLoading, setIsLoading] = useState(false)
 
   useEffect(() => {
-    router.events.on("routeChangeStart", () => setIsLoading(true))
+    router.events.on("routeChangeStart", (url, { shallow }) => {
+      if (!shallow) {
+        setIsLoading(true)
+      }
+    })
     router.events.on("routeChangeComplete", () => setIsLoading(false))
     router.events.on("routeChangeError", () => setIsLoading(false))
   }, [router.events])
diff --git a/dashboard/pages/api/api.ts b/dashboard/pages/api/api.ts
index c5e62cb989378..4164172b89e76 100644
--- a/dashboard/pages/api/api.ts
+++ b/dashboard/pages/api/api.ts
@@ -23,7 +23,9 @@ class Api {
       return data
     } catch (e) {
       console.error(e)
-      throw Error("Failed to fetch " + url)
+      throw Error(`Failed to fetch ${url}`, {
+        cause: e,
+      })
     }
   }
 }
diff --git a/dashboard/pages/api/fetch.ts b/dashboard/pages/api/fetch.ts
index 2ee2bbe9975d1..b545ca60dee25 100644
--- a/dashboard/pages/api/fetch.ts
+++ b/dashboard/pages/api/fetch.ts
@@ -15,12 +15,12 @@
  *
  */
 
-import { useToast } from "@chakra-ui/react"
 import { useEffect, useState } from "react"
+import useErrorToast from "../../hook/useErrorToast"
 
 export default function useFetch<T>(fetchFn: () => Promise<T>) {
   const [response, setResponse] = useState<T>()
-  const toast = useToast()
+  const toast = useErrorToast()
 
   useEffect(() => {
     const fetchData = async () => {
@@ -28,14 +28,7 @@ export default function useFetch<T>(fetchFn: () => Promise<T>) {
         const res = await fetchFn()
         setResponse(res)
       } catch (e: any) {
-        toast({
-          title: "Error Occurred",
-          description: e.toString(),
-          status: "error",
-          duration: 5000,
-          isClosable: true,
-        })
-        console.error(e)
+        toast(e)
       }
     }
     fetchData()
diff --git a/dashboard/pages/await_tree.tsx b/dashboard/pages/await_tree.tsx
index 1be88ab0f0c19..8a3e00d2f89f8 100644
--- a/dashboard/pages/await_tree.tsx
+++ b/dashboard/pages/await_tree.tsx
@@ -81,7 +81,7 @@ export default function AwaitTreeDump() {
 
       result = `${title}\n\n${actorTraces}\n${rpcTraces}`
     } catch (e: any) {
-      result = `${title}\n\nError: ${e.message}`
+      result = `${title}\n\nERROR: ${e.message}\n${e.cause}`
     }
 
     setDump(result)
diff --git a/dashboard/pages/cluster.tsx b/dashboard/pages/cluster.tsx
index 7b5fe1d07684d..5edc80d862dcd 100644
--- a/dashboard/pages/cluster.tsx
+++ b/dashboard/pages/cluster.tsx
@@ -23,7 +23,6 @@ import {
   SimpleGrid,
   Text,
   theme,
-  useToast,
   VStack,
 } from "@chakra-ui/react"
 import { clone, reverse, sortBy } from "lodash"
@@ -32,6 +31,7 @@ import { Fragment, useCallback, useEffect, useState } from "react"
 import { Area, AreaChart, ResponsiveContainer, XAxis, YAxis } from "recharts"
 import { Metrics, MetricsSample } from "../components/metrics"
 import Title from "../components/Title"
+import useErrorToast from "../hook/useErrorToast"
 import { WorkerNode } from "../proto/gen/common"
 import {
   getClusterInfoComputeNode,
@@ -142,7 +142,7 @@ export default function Cluster() {
   const [frontendList, setFrontendList] = useState<WorkerNode[]>([])
   const [computeNodeList, setComputeNodeList] = useState<WorkerNode[]>([])
   const [metrics, setMetrics] = useState<ClusterNodeMetrics>()
-  const toast = useToast()
+  const toast = useErrorToast()
 
   useEffect(() => {
     async function doFetch() {
@@ -150,14 +150,7 @@ export default function Cluster() {
         setFrontendList(await getClusterInfoFrontend())
         setComputeNodeList(await getClusterInfoComputeNode())
       } catch (e: any) {
-        toast({
-          title: "Error Occurred",
-          description: e.toString(),
-          status: "error",
-          duration: 5000,
-          isClosable: true,
-        })
-        console.error(e)
+        toast(e)
       }
     }
     doFetch()
@@ -177,14 +170,7 @@ export default function Cluster() {
           setMetrics(metrics)
           await new Promise((resolve) => setTimeout(resolve, 5000)) // refresh every 5 secs
         } catch (e: any) {
-          toast({
-            title: "Error Occurred",
-            description: e.toString(),
-            status: "error",
-            duration: 5000,
-            isClosable: true,
-          })
-          console.error(e)
+          toast(e, "warning")
           break
         }
       }
diff --git a/dashboard/pages/streaming_graph.tsx b/dashboard/pages/dependency_graph.tsx
similarity index 87%
rename from dashboard/pages/streaming_graph.tsx
rename to dashboard/pages/dependency_graph.tsx
index c2ec4eb731ae1..40a7d4c5c897a 100644
--- a/dashboard/pages/streaming_graph.tsx
+++ b/dashboard/pages/dependency_graph.tsx
@@ -15,7 +15,7 @@
  *
  */
 
-import { Box, Button, Flex, Text, useToast, VStack } from "@chakra-ui/react"
+import { Box, Button, Flex, Text, VStack } from "@chakra-ui/react"
 import { reverse, sortBy } from "lodash"
 import Head from "next/head"
 import Link from "next/link"
@@ -23,8 +23,9 @@ import { useRouter } from "next/router"
 import { Fragment, useCallback, useEffect, useState } from "react"
 import { StreamGraph } from "../components/StreamGraph"
 import Title from "../components/Title"
+import useErrorToast from "../hook/useErrorToast"
 import { ActorPoint } from "../lib/layout"
-import { getRelations, Relation, relationIsStreamingJob } from "./api/streaming"
+import { Relation, getRelations, relationIsStreamingJob } from "./api/streaming"
 
 const SIDEBAR_WIDTH = "200px"
 
@@ -47,7 +48,7 @@ function buildDependencyAsEdges(list: Relation[]): ActorPoint[] {
 }
 
 export default function StreamingGraph() {
-  const toast = useToast()
+  const toast = useErrorToast()
   const [streamingJobList, setStreamingJobList] = useState<Relation[]>()
 
   useEffect(() => {
@@ -55,14 +56,7 @@ export default function StreamingGraph() {
       try {
         setStreamingJobList(await getRelations())
       } catch (e: any) {
-        toast({
-          title: "Error Occurred",
-          description: e.toString(),
-          status: "error",
-          duration: 5000,
-          isClosable: true,
-        })
-        console.error(e)
+        toast(e)
       }
     }
     doFetch()
@@ -83,7 +77,7 @@ export default function StreamingGraph() {
 
   const retVal = (
     <Flex p={3} height="calc(100vh - 20px)" flexDirection="column">
-      <Title>Streaming Graph</Title>
+      <Title>Dependency Graph</Title>
       <Flex flexDirection="row" height="full">
         <Flex
           width={SIDEBAR_WIDTH}
@@ -94,14 +88,14 @@ export default function StreamingGraph() {
           flexDirection="column"
         >
           <Text fontWeight="semibold" mb={3}>
-            All Nodes
+            Relations
           </Text>
           <Box flex={1} overflowY="scroll">
-            <VStack width="full" spacing={1}>
+            <VStack width={SIDEBAR_WIDTH} align="start" spacing={1}>
               {streamingJobList?.map((r) => {
                 const match = router.query.id === r.id.toString()
                 return (
-                  <Link href={`?id=${r.id}`} key={r.id}>
+                  <Link href={`?id=${r.id}`} key={r.id} shallow>
                     <Button
                       colorScheme={match ? "blue" : "gray"}
                       color={match ? "blue.600" : "gray.500"}
diff --git a/dashboard/pages/streaming_plan.tsx b/dashboard/pages/fragment_graph.tsx
similarity index 93%
rename from dashboard/pages/streaming_plan.tsx
rename to dashboard/pages/fragment_graph.tsx
index 3e097275ff641..c5e8accab51be 100644
--- a/dashboard/pages/streaming_plan.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -25,7 +25,6 @@ import {
   Input,
   Select,
   Text,
-  useToast,
   VStack,
 } from "@chakra-ui/react"
 import * as d3 from "d3"
@@ -37,6 +36,7 @@ import { Fragment, useCallback, useEffect, useState } from "react"
 import DependencyGraph from "../components/DependencyGraph"
 import FragmentGraph from "../components/FragmentGraph"
 import Title from "../components/Title"
+import useErrorToast from "../hook/useErrorToast"
 import { ActorBox } from "../lib/layout"
 import { TableFragments, TableFragments_Fragment } from "../proto/gen/meta"
 import { Dispatcher, StreamNode } from "../proto/gen/stream_plan"
@@ -161,16 +161,21 @@ export default function Streaming() {
     return undefined
   }, [fragmentList, router.query.id])
 
+  const setRelationId = useCallback(
+    (id: number) => router.replace(`?id=${id}`, undefined, { shallow: true }),
+    [router]
+  )
+
   useEffect(() => {
     if (relationList) {
       if (!router.query.id) {
         if (relationList.length > 0) {
-          router.replace(`?id=${relationList[0].id}`)
+          setRelationId(relationList[0].id)
         }
       }
     }
     return () => {}
-  }, [router, router.query.id, relationList])
+  }, [router, router.query.id, relationList, setRelationId])
 
   const fragmentDependency = fragmentDependencyCallback()?.fragmentDep
   const fragmentDependencyDag = fragmentDependencyCallback()?.fragmentDepDag
@@ -210,9 +215,7 @@ export default function Streaming() {
   const [searchActorId, setSearchActorId] = useState<string>("")
   const [searchFragId, setSearchFragId] = useState<string>("")
 
-  const setRelationId = (id: number) => router.replace(`?id=${id}`)
-
-  const toast = useToast()
+  const toast = useErrorToast()
 
   const handleSearchFragment = () => {
     const searchFragIdInt = parseInt(searchFragId)
@@ -228,13 +231,7 @@ export default function Streaming() {
       }
     }
 
-    toast({
-      title: "Fragment not found",
-      description: "",
-      status: "error",
-      duration: 5000,
-      isClosable: true,
-    })
+    toast(new Error(`Fragment ${searchFragIdInt} not found`))
   }
 
   const handleSearchActor = () => {
@@ -254,18 +251,12 @@ export default function Streaming() {
       }
     }
 
-    toast({
-      title: "Actor not found",
-      description: "",
-      status: "error",
-      duration: 5000,
-      isClosable: true,
-    })
+    toast(new Error(`Actor ${searchActorIdInt} not found`))
   }
 
   const retVal = (
     <Flex p={3} height="calc(100vh - 20px)" flexDirection="column">
-      <Title>Streaming Plan</Title>
+      <Title>Fragment Graph</Title>
       <Flex flexDirection="row" height="full" width="full">
         <VStack
           mr={3}
@@ -332,12 +323,7 @@ export default function Streaming() {
             </VStack>
           </FormControl>
           <Flex height="full" width="full" flexDirection="column">
-            <Text fontWeight="semibold">Plan</Text>
-            {relationInfo && (
-              <Text>
-                {relationInfo.id} - {relationInfo.name}
-              </Text>
-            )}
+            <Text fontWeight="semibold">Fragments</Text>
             {fragmentDependencyDag && (
               <Box flex="1" overflowY="scroll">
                 <DependencyGraph
diff --git a/e2e_test/batch/catalog/sysinfo.slt.part b/e2e_test/batch/catalog/sysinfo.slt.part
index 0b9520f2e3158..dd6d888b8beb0 100644
--- a/e2e_test/batch/catalog/sysinfo.slt.part
+++ b/e2e_test/batch/catalog/sysinfo.slt.part
@@ -62,6 +62,23 @@ select pg_get_indexdef('tab_idx'::regclass, -1, true), pg_get_indexdef('tab_idx'
 ----
 (empty) (empty)
 
+statement ok
+create materialized view tab_mv as select a.c ac, b.c bc from tab as a join tab as b on a.a = b.b;
+
+query T
+select pg_get_viewdef('tab_mv'::regclass);
+----
+SELECT a.c AS ac, b.c AS bc FROM tab AS a JOIN tab AS b ON a.a = b.b
+
+query error Invalid parameter oid: view or materialized view does not exist:
+select pg_get_viewdef('tab'::regclass);
+
+query error Invalid parameter name: class not found: tab_null
+select pg_get_viewdef('tab_null'::regclass);
+
+statement ok
+drop materialized view tab_mv;
+
 statement ok
 drop index tab_idx;
 
diff --git a/e2e_test/sink/elasticsearch/elasticsearch_sink.result b/e2e_test/sink/elasticsearch/elasticsearch_sink.result
index 7f832258c9595..b67b65495804f 100644
--- a/e2e_test/sink/elasticsearch/elasticsearch_sink.result
+++ b/e2e_test/sink/elasticsearch/elasticsearch_sink.result
@@ -1 +1 @@
-{"took":6,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":6,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"test","_type":"_doc","_id":"3","_score":1.0,"_source":{"d":"1970-01-01","t":"00:00:00.123","tz":"1970-01-01T00:00:00.123Z","v1":3,"v2":2,"v3":"3-2","ts":"1970-01-01 00:00:00.123"}},{"_index":"test","_type":"_doc","_id":"5","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":5,"v2":2,"v3":"5-2","ts":"1970-01-01 00:00:00.000"}},{"_index":"test","_type":"_doc","_id":"8","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":8,"v2":2,"v3":"8-2","ts":"1970-01-01 00:00:00.000"}},{"_index":"test","_type":"_doc","_id":"13","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.123","tz":"1970-01-01T20:00:00.123Z","v1":13,"v2":2,"v3":"13-2","ts":"1970-01-01 20:00:00.123"}},{"_index":"test","_type":"_doc","_id":"1","_score":1.0,"_source":{"d":"2000-01-01","t":"00:00:00.123","tz":"2000-01-01T00:00:00.123Z","v1":1,"v2":50,"v3":"1-50","ts":"2000-01-01 00:00:00.123"}},{"_index":"test","_type":"_doc","_id":"2","_score":1.0,"_source":{"d":"1970-01-01","t":"00:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":2,"v2":2,"v3":"2-2","ts":"1970-01-01 00:00:00.000"}}]}}
+{"took":34,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":6,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"test","_type":"_doc","_id":"2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":2,"st2":2},"t":"00:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":2,"v2":2,"v3":"2-2"}},{"_index":"test","_type":"_doc","_id":"13","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":13,"st2":13},"t":"20:00:00.123456","ts":"1970-01-01 20:00:00.123456","tz":"1970-01-01 20:00:00.123456","v1":13,"v2":2,"v3":"13-2"}},{"_index":"test","_type":"_doc","_id":"3","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":3,"st2":3},"t":"00:00:00.123456","ts":"1970-01-01 00:00:00.123456","tz":"1970-01-01 00:00:00.123456","v1":3,"v2":2,"v3":"3-2"}},{"_index":"test","_type":"_doc","_id":"8","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":8,"st2":8},"t":"20:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":8,"v2":2,"v3":"8-2"}},{"_index":"test","_type":"_doc","_id":"5","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":5,"st2":5},"t":"20:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":5,"v2":2,"v3":"5-2"}},{"_index":"test","_type":"_doc","_id":"1","_score":1.0,"_source":{"d":"2000-01-01","st":{"st1":1,"st2":1},"t":"00:00:00.123456","ts":"2000-01-01 00:00:00.123456","tz":"2000-01-01 00:00:00.123456","v1":1,"v2":50,"v3":"1-50"}}]}}
\ No newline at end of file
diff --git a/e2e_test/sink/elasticsearch/elasticsearch_sink.slt b/e2e_test/sink/elasticsearch/elasticsearch_sink.slt
index 049d111713d6f..70b7e911c0f17 100644
--- a/e2e_test/sink/elasticsearch/elasticsearch_sink.slt
+++ b/e2e_test/sink/elasticsearch/elasticsearch_sink.slt
@@ -6,7 +6,8 @@ CREATE TABLE t7 (
   d date,
   t time,
   ts timestamp,
-  tz timestamptz
+  tz timestamptz,
+  st struct<st1 int, st2 int>,
 );
 
 statement ok
@@ -31,18 +32,18 @@ CREATE SINK s8 from t7 WITH (
 
 statement ok
 INSERT INTO t7 VALUES
-  (1, 2, '1-2', '1970-01-01', '00:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z'),
-  (2, 2, '2-2', '1970-01-01', '00:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z'),
-  (3, 2, '3-2', '1970-01-01', '00:00:00.123456', '1970-01-01 00:00:00.123456', '1970-01-01 00:00:00.123456Z'),
-  (5, 2, '5-2', '1970-01-01', '20:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z'),
-  (8, 2, '8-2', '1970-01-01', '20:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z'),
-  (13, 2, '13-2', '1970-01-01', '20:00:00.123456', '1970-01-01 20:00:00.123456', '1970-01-01 20:00:00.123456Z');
+  (1, 2, '1-2', '1970-01-01', '00:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z',(1,1)),
+  (2, 2, '2-2', '1970-01-01', '00:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z',(2,2)),
+  (3, 2, '3-2', '1970-01-01', '00:00:00.123456', '1970-01-01 00:00:00.123456', '1970-01-01 00:00:00.123456Z',(3,3)),
+  (5, 2, '5-2', '1970-01-01', '20:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z',(5,5)),
+  (8, 2, '8-2', '1970-01-01', '20:00:00', '1970-01-01 00:00:00', '1970-01-01 00:00:00Z',(8,8)),
+  (13, 2, '13-2', '1970-01-01', '20:00:00.123456', '1970-01-01 20:00:00.123456', '1970-01-01 20:00:00.123456Z',(13,13));
 
 statement ok
 FLUSH;
 
 statement ok
-INSERT INTO t7 VALUES (1, 50, '1-50', '2000-01-01', '00:00:00.123456', '2000-01-01 00:00:00.123456', '2000-01-01 00:00:00.123456Z');
+INSERT INTO t7 VALUES (1, 50, '1-50', '2000-01-01', '00:00:00.123456', '2000-01-01 00:00:00.123456', '2000-01-01 00:00:00.123456Z',(1,1));
 
 statement ok
 FLUSH;
diff --git a/e2e_test/sink/elasticsearch/elasticsearch_with_pk_sink.result b/e2e_test/sink/elasticsearch/elasticsearch_with_pk_sink.result
index 12ac1c8370376..3e03f5a511c0a 100644
--- a/e2e_test/sink/elasticsearch/elasticsearch_with_pk_sink.result
+++ b/e2e_test/sink/elasticsearch/elasticsearch_with_pk_sink.result
@@ -1 +1 @@
-{"took":6,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":6,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"test1","_type":"_doc","_id":"3_3-2","_score":1.0,"_source":{"d":"1970-01-01","t":"00:00:00.123","tz":"1970-01-01T00:00:00.123Z","v1":3,"v2":2,"v3":"3-2","ts":"1970-01-01 00:00:00.123"}},{"_index":"test1","_type":"_doc","_id":"5_5-2","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":5,"v2":2,"v3":"5-2","ts":"1970-01-01 00:00:00.000"}},{"_index":"test1","_type":"_doc","_id":"8_8-2","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":8,"v2":2,"v3":"8-2","ts":"1970-01-01 00:00:00.000"}},{"_index":"test1","_type":"_doc","_id":"13_13-2","_score":1.0,"_source":{"d":"1970-01-01","t":"20:00:00.123","tz":"1970-01-01T20:00:00.123Z","v1":13,"v2":2,"v3":"13-2","ts":"1970-01-01 20:00:00.123"}},{"_index":"test1","_type":"_doc","_id":"1_1-50","_score":1.0,"_source":{"d":"2000-01-01","t":"00:00:00.123","tz":"2000-01-01T00:00:00.123Z","v1":1,"v2":50,"v3":"1-50","ts":"2000-01-01 00:00:00.123"}},{"_index":"test1","_type":"_doc","_id":"2_2-2","_score":1.0,"_source":{"d":"1970-01-01","t":"00:00:00.000","tz":"1970-01-01T00:00:00.000Z","v1":2,"v2":2,"v3":"2-2","ts":"1970-01-01 00:00:00.000"}}]}}
+{"took":3,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":6,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"test1","_type":"_doc","_id":"5_5-2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":5,"st2":5},"t":"20:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":5,"v2":2,"v3":"5-2"}},{"_index":"test1","_type":"_doc","_id":"2_2-2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":2,"st2":2},"t":"00:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":2,"v2":2,"v3":"2-2"}},{"_index":"test1","_type":"_doc","_id":"13_13-2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":13,"st2":13},"t":"20:00:00.123456","ts":"1970-01-01 20:00:00.123456","tz":"1970-01-01 20:00:00.123456","v1":13,"v2":2,"v3":"13-2"}},{"_index":"test1","_type":"_doc","_id":"3_3-2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":3,"st2":3},"t":"00:00:00.123456","ts":"1970-01-01 00:00:00.123456","tz":"1970-01-01 00:00:00.123456","v1":3,"v2":2,"v3":"3-2"}},{"_index":"test1","_type":"_doc","_id":"8_8-2","_score":1.0,"_source":{"d":"1970-01-01","st":{"st1":8,"st2":8},"t":"20:00:00.000000","ts":"1970-01-01 00:00:00.000000","tz":"1970-01-01 00:00:00.000000","v1":8,"v2":2,"v3":"8-2"}},{"_index":"test1","_type":"_doc","_id":"1_1-50","_score":1.0,"_source":{"d":"2000-01-01","st":{"st1":1,"st2":1},"t":"00:00:00.123456","ts":"2000-01-01 00:00:00.123456","tz":"2000-01-01 00:00:00.123456","v1":1,"v2":50,"v3":"1-50"}}]}}
\ No newline at end of file
diff --git a/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkWriterStreamObserver.java b/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkWriterStreamObserver.java
index a8b53f5c7e481..cd61da38d6cb5 100644
--- a/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkWriterStreamObserver.java
+++ b/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkWriterStreamObserver.java
@@ -101,7 +101,7 @@ public void onNext(ConnectorServiceProto.SinkWriterStreamRequest sinkTask) {
                             .asRuntimeException();
                 }
                 sinkId = sinkTask.getStart().getSinkParam().getSinkId();
-                bindSink(sinkTask.getStart().getSinkParam(), sinkTask.getStart().getFormat());
+                bindSink(sinkTask.getStart());
                 currentEpoch = null;
                 currentBatchId = null;
                 epochStarted = false;
@@ -200,14 +200,13 @@ private void cleanup() {
         ConnectorNodeMetrics.decActiveSinkConnections(connectorName, "node1");
     }
 
-    private void bindSink(
-            ConnectorServiceProto.SinkParam sinkParam,
-            ConnectorServiceProto.SinkPayloadFormat format) {
-        tableSchema = TableSchema.fromProto(sinkParam.getTableSchema());
+    private void bindSink(ConnectorServiceProto.SinkWriterStreamRequest.StartSink startSink) {
+        var sinkParam = startSink.getSinkParam();
+        tableSchema = TableSchema.fromProto(startSink.getPayloadSchema());
         String connectorName = getConnectorName(sinkParam);
         SinkFactory sinkFactory = SinkUtils.getSinkFactory(connectorName);
         sink = sinkFactory.createWriter(tableSchema, sinkParam.getPropertiesMap());
-        switch (format) {
+        switch (startSink.getFormat()) {
             case FORMAT_UNSPECIFIED:
             case UNRECOGNIZED:
                 throw INVALID_ARGUMENT
diff --git a/java/connector-node/risingwave-connector-test/src/test/java/com/risingwave/connector/sink/elasticsearch/EsSinkTest.java b/java/connector-node/risingwave-connector-test/src/test/java/com/risingwave/connector/sink/elasticsearch/EsSinkTest.java
index 8a61c4e8fba07..e7e74de2af6f0 100644
--- a/java/connector-node/risingwave-connector-test/src/test/java/com/risingwave/connector/sink/elasticsearch/EsSinkTest.java
+++ b/java/connector-node/risingwave-connector-test/src/test/java/com/risingwave/connector/sink/elasticsearch/EsSinkTest.java
@@ -61,8 +61,8 @@ public void testEsSink(ElasticsearchContainer container, String username, String
                         getTestTableSchema());
         sink.write(
                 Iterators.forArray(
-                        new ArraySinkRow(Op.INSERT, 1, "Alice"),
-                        new ArraySinkRow(Op.INSERT, 2, "Bob")));
+                        new ArraySinkRow(Op.INSERT, "1$Alice", "{\"id\":1,\"name\":\"Alice\"}"),
+                        new ArraySinkRow(Op.INSERT, "2$Bob", "{\"id\":2,\"name\":\"Bob\"}")));
         sink.sync();
         // container is slow here, but our default flush time is 5s,
         // so 3s is enough for sync test
diff --git a/java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSink.java b/java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSink.java
index dd579aafb6035..7a2e5742bb790 100644
--- a/java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSink.java
+++ b/java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSink.java
@@ -16,8 +16,6 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.risingwave.connector.api.TableSchema;
 import com.risingwave.connector.api.sink.SinkRow;
 import com.risingwave.connector.api.sink.SinkWriterBase;
@@ -27,7 +25,6 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.UsernamePasswordCredentials;
@@ -47,6 +44,7 @@
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.xcontent.XContentType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -67,11 +65,6 @@ public class EsSink extends SinkWriterBase {
     private static final Logger LOG = LoggerFactory.getLogger(EsSink.class);
     private static final String ERROR_REPORT_TEMPLATE = "Error message %s";
 
-    private static final TimeZone UTCTimeZone = TimeZone.getTimeZone("UTC");
-    private final SimpleDateFormat tDfm;
-    private final SimpleDateFormat tsDfm;
-    private final SimpleDateFormat tstzDfm;
-
     private final EsSinkConfig config;
     private BulkProcessor bulkProcessor;
     private final RestHighLevelClient client;
@@ -79,9 +72,6 @@ public class EsSink extends SinkWriterBase {
     // Used to handle the return message of ES and throw errors
     private final RequestTracker requestTracker;
 
-    // For bulk listener
-    private final List<Integer> primaryKeyIndexes;
-
     class RequestTracker {
         // Used to save the return results of es asynchronous writes. The capacity is Integer.Max
         private final BlockingQueue<EsWriteResultResp> blockingQueue = new LinkedBlockingQueue<>();
@@ -196,15 +186,6 @@ public EsSink(EsSinkConfig config, TableSchema tableSchema) {
             throw Status.INTERNAL.withDescription(e.getMessage()).asRuntimeException();
         }
         this.bulkProcessor = createBulkProcessor(this.requestTracker);
-
-        primaryKeyIndexes = new ArrayList<Integer>();
-        for (String primaryKey : tableSchema.getPrimaryKeys()) {
-            primaryKeyIndexes.add(tableSchema.getColumnIndex(primaryKey));
-        }
-
-        tDfm = createSimpleDateFormat("HH:mm:ss.SSS", UTCTimeZone);
-        tsDfm = createSimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS", UTCTimeZone);
-        tstzDfm = createSimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", UTCTimeZone);
     }
 
     private static RestClientBuilder configureRestClientBuilder(
@@ -297,116 +278,20 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
         }
     }
 
-    /**
-     * The api accepts doc in map form.
-     *
-     * @param row
-     * @return Map from Field name to Value
-     * @throws JsonProcessingException
-     * @throws JsonMappingException
-     */
-    private Map<String, Object> buildDoc(SinkRow row)
-            throws JsonMappingException, JsonProcessingException {
-        Map<String, Object> doc = new HashMap();
-        var tableSchema = getTableSchema();
-        var columnDescs = tableSchema.getColumnDescs();
-        for (int i = 0; i < row.size(); i++) {
-            var type = columnDescs.get(i).getDataType().getTypeName();
-            Object col = row.get(i);
-            switch (type) {
-                    // es client doesn't natively support java.sql.Timestamp/Time/Date
-                    // so we need to convert Date/Time/Timestamp type into a string as suggested in
-                    // https://github.com/elastic/elasticsearch/issues/31377#issuecomment-398102292
-                case DATE:
-                    col = col.toString();
-                    break;
-                    // construct java.sql.Time/Timestamp with milliseconds time value.
-                    // it will use system timezone by default, so we have to set timezone manually
-                case TIME:
-                    col = tDfm.format(col);
-                    break;
-                case TIMESTAMP:
-                    col = tsDfm.format(col);
-                    break;
-                case TIMESTAMPTZ:
-                    col = tstzDfm.format(col);
-                    break;
-                case JSONB:
-                    ObjectMapper mapper = new ObjectMapper();
-                    JsonNode jsonNode = mapper.readTree((String) col);
-                    col = convertJsonNode(jsonNode);
-                    break;
-                default:
-                    break;
-            }
-
-            doc.put(getTableSchema().getColumnDesc(i).getName(), col);
-        }
-        return doc;
-    }
-
-    private static Object convertJsonNode(JsonNode jsonNode) {
-        if (jsonNode.isObject()) {
-            Map<String, Object> resultMap = new HashMap<>();
-            jsonNode.fields()
-                    .forEachRemaining(
-                            entry -> {
-                                resultMap.put(entry.getKey(), convertJsonNode(entry.getValue()));
-                            });
-            return resultMap;
-        } else if (jsonNode.isArray()) {
-            List<Object> resultList = new ArrayList<>();
-            jsonNode.elements()
-                    .forEachRemaining(
-                            element -> {
-                                resultList.add(convertJsonNode(element));
-                            });
-            return resultList;
-        } else if (jsonNode.isNumber()) {
-            return jsonNode.numberValue();
-        } else if (jsonNode.isTextual()) {
-            return jsonNode.textValue();
-        } else if (jsonNode.isBoolean()) {
-            return jsonNode.booleanValue();
-        } else if (jsonNode.isNull()) {
-            return null;
-        } else {
-            throw new IllegalArgumentException("Unsupported JSON type");
-        }
-    }
-
-    /**
-     * use primary keys as id concatenated by a specific delimiter.
-     *
-     * @param row
-     * @return
-     */
-    private String buildId(SinkRow row) {
-        String id;
-        if (primaryKeyIndexes.isEmpty()) {
-            id = row.get(0).toString();
-        } else {
-            List<String> keys =
-                    primaryKeyIndexes.stream()
-                            .map(index -> row.get(index).toString())
-                            .collect(Collectors.toList());
-            id = String.join(config.getDelimiter(), keys);
-        }
-        return id;
-    }
-
     private void processUpsert(SinkRow row) throws JsonMappingException, JsonProcessingException {
-        Map<String, Object> doc = buildDoc(row);
-        final String key = buildId(row);
+        final String key = (String) row.get(0);
+        String doc = (String) row.get(1);
 
         UpdateRequest updateRequest =
-                new UpdateRequest(config.getIndex(), "_doc", key).doc(doc).upsert(doc);
+                new UpdateRequest(config.getIndex(), "_doc", key).doc(doc, XContentType.JSON);
+        updateRequest.docAsUpsert(true);
         this.requestTracker.addWriteTask();
         bulkProcessor.add(updateRequest);
     }
 
-    private void processDelete(SinkRow row) {
-        final String key = buildId(row);
+    private void processDelete(SinkRow row) throws JsonMappingException, JsonProcessingException {
+        final String key = (String) row.get(0);
+
         DeleteRequest deleteRequest = new DeleteRequest(config.getIndex(), "_doc", key);
         this.requestTracker.addWriteTask();
         bulkProcessor.add(deleteRequest);
diff --git a/proto/connector_service.proto b/proto/connector_service.proto
index ddda62c6aace5..465af0d2a55a8 100644
--- a/proto/connector_service.proto
+++ b/proto/connector_service.proto
@@ -39,6 +39,7 @@ message SinkWriterStreamRequest {
   message StartSink {
     SinkParam sink_param = 1;
     SinkPayloadFormat format = 2;
+    TableSchema payload_schema = 3;
   }
 
   message WriteBatch {
diff --git a/proto/expr.proto b/proto/expr.proto
index 4b195183c4df2..31835ac905705 100644
--- a/proto/expr.proto
+++ b/proto/expr.proto
@@ -274,6 +274,7 @@ message ExprNode {
     // System information functions
     PG_GET_INDEXDEF = 2400;
     COL_DESCRIPTION = 2401;
+    PG_GET_VIEWDEF = 2402;
   }
   Type function_type = 1;
   data.DataType return_type = 3;
diff --git a/src/connector/src/sink/big_query.rs b/src/connector/src/sink/big_query.rs
index 081b5814b01e3..41cfe7783819f 100644
--- a/src/connector/src/sink/big_query.rs
+++ b/src/connector/src/sink/big_query.rs
@@ -33,7 +33,10 @@ use url::Url;
 use with_options::WithOptions;
 use yup_oauth2::ServiceAccountKey;
 
-use super::encoder::{JsonEncoder, RowEncoder, TimestampHandlingMode};
+use super::encoder::{
+    DateHandlingMode, JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode,
+    TimestamptzHandlingMode,
+};
 use super::writer::LogSinkerOf;
 use super::{SinkError, SINK_TYPE_APPEND_ONLY, SINK_TYPE_OPTION, SINK_TYPE_UPSERT};
 use crate::aws_utils::load_file_descriptor_from_s3;
@@ -309,10 +312,13 @@ impl BigQuerySinkWriter {
             client,
             is_append_only,
             insert_request: TableDataInsertAllRequest::new(),
-            row_encoder: JsonEncoder::new_with_big_query(
+            row_encoder: JsonEncoder::new(
                 schema,
                 None,
+                DateHandlingMode::String,
                 TimestampHandlingMode::String,
+                TimestamptzHandlingMode::UtcString,
+                TimeHandlingMode::Milli,
             ),
         })
     }
diff --git a/src/connector/src/sink/elasticsearch.rs b/src/connector/src/sink/elasticsearch.rs
new file mode 100644
index 0000000000000..f6ddbeb7e35d2
--- /dev/null
+++ b/src/connector/src/sink/elasticsearch.rs
@@ -0,0 +1,138 @@
+// Copyright 2024 RisingWave Labs
+//
+// 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.
+
+use std::collections::HashMap;
+
+use anyhow::anyhow;
+use risingwave_common::array::{
+    ArrayImpl, JsonbArrayBuilder, RowRef, StreamChunk, Utf8ArrayBuilder,
+};
+use risingwave_common::catalog::Schema;
+use risingwave_common::row::Row;
+use risingwave_common::types::{JsonbVal, Scalar, ToText};
+use serde_json::Value;
+
+use super::encoder::{JsonEncoder, RowEncoder};
+use super::remote::ElasticSearchSink;
+use crate::sink::{Result, Sink};
+pub const ES_OPTION_DELIMITER: &str = "delimiter";
+
+pub enum StreamChunkConverter {
+    Es(EsStreamChunkConverter),
+    Other,
+}
+impl StreamChunkConverter {
+    pub fn new(
+        sink_name: &str,
+        schema: Schema,
+        pk_indices: &Vec<usize>,
+        properties: &HashMap<String, String>,
+    ) -> Result<Self> {
+        if sink_name == ElasticSearchSink::SINK_NAME {
+            Ok(StreamChunkConverter::Es(EsStreamChunkConverter::new(
+                schema,
+                pk_indices.clone(),
+                properties.get(ES_OPTION_DELIMITER).cloned(),
+            )?))
+        } else {
+            Ok(StreamChunkConverter::Other)
+        }
+    }
+
+    pub fn convert_chunk(&self, chunk: StreamChunk) -> Result<StreamChunk> {
+        match self {
+            StreamChunkConverter::Es(es) => es.convert_chunk(chunk),
+            StreamChunkConverter::Other => Ok(chunk),
+        }
+    }
+}
+pub struct EsStreamChunkConverter {
+    json_encoder: JsonEncoder,
+    fn_build_id: Box<dyn Fn(RowRef<'_>) -> Result<String> + Send>,
+}
+impl EsStreamChunkConverter {
+    fn new(schema: Schema, pk_indices: Vec<usize>, delimiter: Option<String>) -> Result<Self> {
+        let fn_build_id: Box<dyn Fn(RowRef<'_>) -> Result<String> + Send> = if pk_indices.is_empty()
+        {
+            Box::new(|row: RowRef<'_>| {
+                Ok(row
+                    .datum_at(0)
+                    .ok_or_else(|| anyhow!("No value find in row, index is 0"))?
+                    .to_text())
+            })
+        } else if pk_indices.len() == 1 {
+            let index = *pk_indices.get(0).unwrap();
+            Box::new(move |row: RowRef<'_>| {
+                Ok(row
+                    .datum_at(index)
+                    .ok_or_else(|| anyhow!("No value find in row, index is 0"))?
+                    .to_text())
+            })
+        } else {
+            let delimiter = delimiter
+                .as_ref()
+                .ok_or_else(|| anyhow!("Please set delimiter in with option"))?
+                .clone();
+            Box::new(move |row: RowRef<'_>| {
+                let mut keys = vec![];
+                for index in &pk_indices {
+                    keys.push(
+                        row.datum_at(*index)
+                            .ok_or_else(|| anyhow!("No value find in row, index is {}", index))?
+                            .to_text(),
+                    );
+                }
+                Ok(keys.join(&delimiter))
+            })
+        };
+        let json_encoder = JsonEncoder::new_with_es(schema, None);
+        Ok(Self {
+            json_encoder,
+            fn_build_id,
+        })
+    }
+
+    fn convert_chunk(&self, chunk: StreamChunk) -> Result<StreamChunk> {
+        let mut ops = vec![];
+        let mut id_string_builder =
+            <Utf8ArrayBuilder as risingwave_common::array::ArrayBuilder>::new(chunk.capacity());
+        let mut json_builder =
+            <JsonbArrayBuilder as risingwave_common::array::ArrayBuilder>::new(chunk.capacity());
+        for (op, row) in chunk.rows() {
+            ops.push(op);
+            let json = JsonbVal::from(Value::Object(self.json_encoder.encode(row)?));
+            risingwave_common::array::ArrayBuilder::append(
+                &mut id_string_builder,
+                Some(&self.build_id(row)?),
+            );
+            risingwave_common::array::ArrayBuilder::append(
+                &mut json_builder,
+                Some(json.as_scalar_ref()),
+            );
+        }
+        let json_array = risingwave_common::array::ArrayBuilder::finish(json_builder);
+        let id_string_array = risingwave_common::array::ArrayBuilder::finish(id_string_builder);
+        Ok(StreamChunk::new(
+            ops,
+            vec![
+                std::sync::Arc::new(ArrayImpl::Utf8(id_string_array)),
+                std::sync::Arc::new(ArrayImpl::Jsonb(json_array)),
+            ],
+        ))
+    }
+
+    fn build_id(&self, row: RowRef<'_>) -> Result<String> {
+        (self.fn_build_id)(row)
+    }
+}
diff --git a/src/connector/src/sink/encoder/json.rs b/src/connector/src/sink/encoder/json.rs
index 473db379341b5..b4d2de84c0069 100644
--- a/src/connector/src/sink/encoder/json.rs
+++ b/src/connector/src/sink/encoder/json.rs
@@ -23,19 +23,20 @@ use itertools::Itertools;
 use risingwave_common::array::{ArrayError, ArrayResult};
 use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::row::Row;
-use risingwave_common::types::{DataType, DatumRef, Decimal, ScalarRefImpl, ToText};
+use risingwave_common::types::{DataType, DatumRef, Decimal, JsonbVal, ScalarRefImpl, ToText};
 use risingwave_common::util::iter_util::ZipEqDebug;
 use serde_json::{json, Map, Value};
 
 use super::{
     CustomJsonType, DateHandlingMode, KafkaConnectParams, KafkaConnectParamsRef, Result,
-    RowEncoder, SerTo, TimestampHandlingMode, TimestamptzHandlingMode,
+    RowEncoder, SerTo, TimeHandlingMode, TimestampHandlingMode, TimestamptzHandlingMode,
 };
 use crate::sink::SinkError;
 
 pub struct JsonEncoder {
     schema: Schema,
     col_indices: Option<Vec<usize>>,
+    time_handling_mode: TimeHandlingMode,
     date_handling_mode: DateHandlingMode,
     timestamp_handling_mode: TimestampHandlingMode,
     timestamptz_handling_mode: TimestamptzHandlingMode,
@@ -50,10 +51,12 @@ impl JsonEncoder {
         date_handling_mode: DateHandlingMode,
         timestamp_handling_mode: TimestampHandlingMode,
         timestamptz_handling_mode: TimestamptzHandlingMode,
+        time_handling_mode: TimeHandlingMode,
     ) -> Self {
         Self {
             schema,
             col_indices,
+            time_handling_mode,
             date_handling_mode,
             timestamp_handling_mode,
             timestamptz_handling_mode,
@@ -62,6 +65,19 @@ impl JsonEncoder {
         }
     }
 
+    pub fn new_with_es(schema: Schema, col_indices: Option<Vec<usize>>) -> Self {
+        Self {
+            schema,
+            col_indices,
+            time_handling_mode: TimeHandlingMode::String,
+            date_handling_mode: DateHandlingMode::String,
+            timestamp_handling_mode: TimestampHandlingMode::String,
+            timestamptz_handling_mode: TimestamptzHandlingMode::UtcWithoutSuffix,
+            custom_json_type: CustomJsonType::Es,
+            kafka_connect: None,
+        }
+    }
+
     pub fn new_with_doris(
         schema: Schema,
         col_indices: Option<Vec<usize>>,
@@ -71,6 +87,7 @@ impl JsonEncoder {
         Self {
             schema,
             col_indices,
+            time_handling_mode: TimeHandlingMode::Milli,
             date_handling_mode: DateHandlingMode::String,
             timestamp_handling_mode,
             timestamptz_handling_mode: TimestamptzHandlingMode::UtcWithoutSuffix,
@@ -85,22 +102,6 @@ impl JsonEncoder {
             ..self
         }
     }
-
-    pub fn new_with_big_query(
-        schema: Schema,
-        col_indices: Option<Vec<usize>>,
-        timestamp_handling_mode: TimestampHandlingMode,
-    ) -> Self {
-        Self {
-            schema,
-            col_indices,
-            date_handling_mode: DateHandlingMode::String,
-            timestamp_handling_mode,
-            timestamptz_handling_mode: TimestamptzHandlingMode::UtcString,
-            custom_json_type: CustomJsonType::Bigquery,
-            kafka_connect: None,
-        }
-    }
 }
 
 impl RowEncoder for JsonEncoder {
@@ -130,6 +131,7 @@ impl RowEncoder for JsonEncoder {
                 self.date_handling_mode,
                 self.timestamp_handling_mode,
                 self.timestamptz_handling_mode,
+                self.time_handling_mode,
                 &self.custom_json_type,
             )
             .map_err(|e| SinkError::Encode(e.to_string()))?;
@@ -166,6 +168,7 @@ fn datum_to_json_object(
     date_handling_mode: DateHandlingMode,
     timestamp_handling_mode: TimestampHandlingMode,
     timestamptz_handling_mode: TimestamptzHandlingMode,
+    time_handling_mode: TimeHandlingMode,
     custom_json_type: &CustomJsonType,
 ) -> ArrayResult<Value> {
     let scalar_ref = match datum {
@@ -216,7 +219,7 @@ fn datum_to_json_object(
                 }
                 json!(v_string)
             }
-            CustomJsonType::None | CustomJsonType::Bigquery => {
+            CustomJsonType::Es | CustomJsonType::None => {
                 json!(v.to_text())
             }
         },
@@ -234,10 +237,16 @@ fn datum_to_json_object(
             TimestamptzHandlingMode::Micro => json!(v.timestamp_micros()),
             TimestamptzHandlingMode::Milli => json!(v.timestamp_millis()),
         },
-        (DataType::Time, ScalarRefImpl::Time(v)) => {
-            // todo: just ignore the nanos part to avoid leap second complex
-            json!(v.0.num_seconds_from_midnight() as i64 * 1000)
-        }
+        (DataType::Time, ScalarRefImpl::Time(v)) => match time_handling_mode {
+            TimeHandlingMode::Milli => {
+                // todo: just ignore the nanos part to avoid leap second complex
+                json!(v.0.num_seconds_from_midnight() as i64 * 1000)
+            }
+            TimeHandlingMode::String => {
+                let a = v.0.format("%H:%M:%S%.6f").to_string();
+                json!(a)
+            }
+        },
         (DataType::Date, ScalarRefImpl::Date(v)) => match date_handling_mode {
             DateHandlingMode::FromCe => json!(v.0.num_days_from_ce()),
             DateHandlingMode::FromEpoch => {
@@ -260,9 +269,10 @@ fn datum_to_json_object(
         (DataType::Interval, ScalarRefImpl::Interval(v)) => {
             json!(v.as_iso_8601())
         }
-        (DataType::Jsonb, ScalarRefImpl::Jsonb(jsonb_ref)) => {
-            json!(jsonb_ref.to_string())
-        }
+        (DataType::Jsonb, ScalarRefImpl::Jsonb(jsonb_ref)) => match custom_json_type {
+            CustomJsonType::Es => JsonbVal::from(jsonb_ref).take(),
+            CustomJsonType::Doris(_) | CustomJsonType::None => json!(jsonb_ref.to_string()),
+        },
         (DataType::List(datatype), ScalarRefImpl::List(list_ref)) => {
             let elems = list_ref.iter();
             let mut vec = Vec::with_capacity(elems.len());
@@ -274,6 +284,7 @@ fn datum_to_json_object(
                     date_handling_mode,
                     timestamp_handling_mode,
                     timestamptz_handling_mode,
+                    time_handling_mode,
                     custom_json_type,
                 )?;
                 vec.push(value);
@@ -295,6 +306,7 @@ fn datum_to_json_object(
                             date_handling_mode,
                             timestamp_handling_mode,
                             timestamptz_handling_mode,
+                            time_handling_mode,
                             custom_json_type,
                         )?;
                         map.insert(sub_field.name.clone(), value);
@@ -303,7 +315,7 @@ fn datum_to_json_object(
                         ArrayError::internal(format!("Json to string err{:?}", err))
                     })?)
                 }
-                CustomJsonType::None | CustomJsonType::Bigquery => {
+                CustomJsonType::Es | CustomJsonType::None => {
                     let mut map = Map::with_capacity(st.len());
                     for (sub_datum_ref, sub_field) in struct_ref.iter_fields_ref().zip_eq_debug(
                         st.iter()
@@ -315,6 +327,7 @@ fn datum_to_json_object(
                             date_handling_mode,
                             timestamp_handling_mode,
                             timestamptz_handling_mode,
+                            time_handling_mode,
                             custom_json_type,
                         )?;
                         map.insert(sub_field.name.clone(), value);
@@ -433,6 +446,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -447,6 +461,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -461,6 +476,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -480,6 +496,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -495,6 +512,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcWithoutSuffix,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -512,6 +530,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::Milli,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -529,6 +548,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -547,6 +567,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -564,6 +585,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -581,6 +603,7 @@ mod tests {
             DateHandlingMode::String,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::Doris(map),
         )
         .unwrap();
@@ -595,6 +618,7 @@ mod tests {
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -609,6 +633,7 @@ mod tests {
             DateHandlingMode::FromEpoch,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::None,
         )
         .unwrap();
@@ -623,6 +648,7 @@ mod tests {
             DateHandlingMode::String,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::Doris(HashMap::default()),
         )
         .unwrap();
@@ -647,6 +673,7 @@ mod tests {
             DateHandlingMode::String,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
             &CustomJsonType::Doris(HashMap::default()),
         )
         .unwrap();
diff --git a/src/connector/src/sink/encoder/mod.rs b/src/connector/src/sink/encoder/mod.rs
index ad32e78aa48ce..83c28ce4b4a5a 100644
--- a/src/connector/src/sink/encoder/mod.rs
+++ b/src/connector/src/sink/encoder/mod.rs
@@ -97,6 +97,12 @@ pub enum TimestampHandlingMode {
     String,
 }
 
+#[derive(Clone, Copy)]
+pub enum TimeHandlingMode {
+    Milli,
+    String,
+}
+
 #[derive(Clone, Copy, Default)]
 pub enum TimestamptzHandlingMode {
     #[default]
@@ -134,8 +140,8 @@ pub enum CustomJsonType {
     // The internal order of the struct should follow the insertion order.
     // The decimal needs verification and calibration.
     Doris(HashMap<String, (u8, u8)>),
-    // Bigquery's json need date is string.
-    Bigquery,
+    // Es's json need jsonb is struct
+    Es,
     None,
 }
 
diff --git a/src/connector/src/sink/formatter/debezium_json.rs b/src/connector/src/sink/formatter/debezium_json.rs
index ae98e7b51e62f..6fff15058bd6f 100644
--- a/src/connector/src/sink/formatter/debezium_json.rs
+++ b/src/connector/src/sink/formatter/debezium_json.rs
@@ -21,7 +21,8 @@ use tracing::warn;
 
 use super::{Result, SinkFormatter, StreamChunk};
 use crate::sink::encoder::{
-    DateHandlingMode, JsonEncoder, RowEncoder, TimestampHandlingMode, TimestamptzHandlingMode,
+    DateHandlingMode, JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode,
+    TimestamptzHandlingMode,
 };
 use crate::tri;
 
@@ -67,6 +68,7 @@ impl DebeziumJsonFormatter {
             DateHandlingMode::FromEpoch,
             TimestampHandlingMode::Milli,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
         );
         let val_encoder = JsonEncoder::new(
             schema.clone(),
@@ -74,6 +76,7 @@ impl DebeziumJsonFormatter {
             DateHandlingMode::FromEpoch,
             TimestampHandlingMode::Milli,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
         );
         Self {
             schema,
@@ -393,6 +396,7 @@ mod tests {
             DateHandlingMode::FromEpoch,
             TimestampHandlingMode::Milli,
             TimestamptzHandlingMode::UtcString,
+            TimeHandlingMode::Milli,
         );
         let json_chunk = chunk_to_json(chunk, &encoder).unwrap();
         let schema_json = schema_to_json(&schema, "test_db", "test_table");
diff --git a/src/connector/src/sink/formatter/mod.rs b/src/connector/src/sink/formatter/mod.rs
index 91025bf3b401b..74c6d54151133 100644
--- a/src/connector/src/sink/formatter/mod.rs
+++ b/src/connector/src/sink/formatter/mod.rs
@@ -28,7 +28,9 @@ pub use upsert::UpsertFormatter;
 
 use super::catalog::{SinkEncode, SinkFormat, SinkFormatDesc};
 use super::encoder::template::TemplateEncoder;
-use super::encoder::{DateHandlingMode, KafkaConnectParams, TimestamptzHandlingMode};
+use super::encoder::{
+    DateHandlingMode, KafkaConnectParams, TimeHandlingMode, TimestamptzHandlingMode,
+};
 use super::redis::{KEY_FORMAT, VALUE_FORMAT};
 use crate::sink::encoder::{
     AvroEncoder, AvroHeader, JsonEncoder, ProtoEncoder, TimestampHandlingMode,
@@ -102,6 +104,7 @@ impl SinkFormatterImpl {
                         DateHandlingMode::FromCe,
                         TimestampHandlingMode::Milli,
                         timestamptz_mode,
+                        TimeHandlingMode::Milli,
                     )
                 });
 
@@ -113,6 +116,7 @@ impl SinkFormatterImpl {
                             DateHandlingMode::FromCe,
                             TimestampHandlingMode::Milli,
                             timestamptz_mode,
+                            TimeHandlingMode::Milli,
                         );
                         let formatter = AppendOnlyFormatter::new(key_encoder, val_encoder);
                         Ok(SinkFormatterImpl::AppendOnlyJson(formatter))
@@ -174,6 +178,7 @@ impl SinkFormatterImpl {
                             DateHandlingMode::FromCe,
                             TimestampHandlingMode::Milli,
                             timestamptz_mode,
+                            TimeHandlingMode::Milli,
                         );
                         let mut val_encoder = JsonEncoder::new(
                             schema,
@@ -181,6 +186,7 @@ impl SinkFormatterImpl {
                             DateHandlingMode::FromCe,
                             TimestampHandlingMode::Milli,
                             timestamptz_mode,
+                            TimeHandlingMode::Milli,
                         );
 
                         if let Some(s) = format_desc.options.get("schemas.enable") {
diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs
index fd9935a12f5d5..0c59780916cb4 100644
--- a/src/connector/src/sink/kafka.rs
+++ b/src/connector/src/sink/kafka.rs
@@ -565,7 +565,8 @@ mod test {
 
     use super::*;
     use crate::sink::encoder::{
-        DateHandlingMode, JsonEncoder, TimestampHandlingMode, TimestamptzHandlingMode,
+        DateHandlingMode, JsonEncoder, TimeHandlingMode, TimestampHandlingMode,
+        TimestamptzHandlingMode,
     };
     use crate::sink::formatter::AppendOnlyFormatter;
 
@@ -738,6 +739,7 @@ mod test {
                     DateHandlingMode::FromCe,
                     TimestampHandlingMode::Milli,
                     TimestamptzHandlingMode::UtcString,
+                    TimeHandlingMode::Milli,
                 ),
             )),
         )
diff --git a/src/connector/src/sink/mod.rs b/src/connector/src/sink/mod.rs
index 7bbd42edcd338..e5f9df155c735 100644
--- a/src/connector/src/sink/mod.rs
+++ b/src/connector/src/sink/mod.rs
@@ -20,6 +20,7 @@ pub mod coordinate;
 pub mod deltalake;
 pub mod doris;
 pub mod doris_starrocks_connector;
+pub mod elasticsearch;
 pub mod encoder;
 pub mod formatter;
 pub mod iceberg;
diff --git a/src/connector/src/sink/nats.rs b/src/connector/src/sink/nats.rs
index a51ab42050fd2..01bac5e6b048c 100644
--- a/src/connector/src/sink/nats.rs
+++ b/src/connector/src/sink/nats.rs
@@ -25,7 +25,7 @@ use tokio_retry::strategy::{jitter, ExponentialBackoff};
 use tokio_retry::Retry;
 use with_options::WithOptions;
 
-use super::encoder::{DateHandlingMode, TimestamptzHandlingMode};
+use super::encoder::{DateHandlingMode, TimeHandlingMode, TimestamptzHandlingMode};
 use super::utils::chunk_to_json;
 use super::{DummySinkCommitCoordinator, SinkWriterParam};
 use crate::common::NatsCommon;
@@ -146,6 +146,7 @@ impl NatsSinkWriter {
                 DateHandlingMode::FromCe,
                 TimestampHandlingMode::Milli,
                 TimestamptzHandlingMode::UtcWithoutSuffix,
+                TimeHandlingMode::Milli,
             ),
         })
     }
diff --git a/src/connector/src/sink/remote.rs b/src/connector/src/sink/remote.rs
index f76b685f96c17..320f77c6a47ba 100644
--- a/src/connector/src/sink/remote.rs
+++ b/src/connector/src/sink/remote.rs
@@ -26,6 +26,7 @@ use itertools::Itertools;
 use jni::JavaVM;
 use prost::Message;
 use risingwave_common::array::StreamChunk;
+use risingwave_common::catalog::{ColumnDesc, ColumnId};
 use risingwave_common::error::anyhow_error;
 use risingwave_common::types::DataType;
 use risingwave_common::util::drop_either_future;
@@ -39,8 +40,9 @@ use risingwave_pb::connector_service::sink_writer_stream_request::{
 };
 use risingwave_pb::connector_service::{
     sink_coordinator_stream_request, sink_coordinator_stream_response, sink_writer_stream_response,
-    SinkCoordinatorStreamRequest, SinkCoordinatorStreamResponse, SinkMetadata, SinkPayloadFormat,
-    SinkWriterStreamRequest, SinkWriterStreamResponse, ValidateSinkRequest, ValidateSinkResponse,
+    PbSinkParam, SinkCoordinatorStreamRequest, SinkCoordinatorStreamResponse, SinkMetadata,
+    SinkPayloadFormat, SinkWriterStreamRequest, SinkWriterStreamResponse, TableSchema,
+    ValidateSinkRequest, ValidateSinkResponse,
 };
 use risingwave_rpc_client::error::RpcError;
 use risingwave_rpc_client::{
@@ -53,6 +55,7 @@ use tokio::task::spawn_blocking;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::warn;
 
+use super::elasticsearch::{StreamChunkConverter, ES_OPTION_DELIMITER};
 use crate::sink::catalog::desc::SinkDesc;
 use crate::sink::coordinate::CoordinatedSinkWriter;
 use crate::sink::log_store::{LogStoreReadItem, TruncateOffset};
@@ -145,21 +148,28 @@ impl<R: RemoteSinkTrait> Sink for RemoteSink<R> {
     }
 
     async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
-        RemoteLogSinker::new(self.param.clone(), writer_param).await
+        RemoteLogSinker::new(self.param.clone(), writer_param, Self::SINK_NAME).await
     }
 
     async fn validate(&self) -> Result<()> {
-        validate_remote_sink(&self.param).await?;
+        validate_remote_sink(&self.param, Self::SINK_NAME).await?;
         Ok(())
     }
 }
 
-async fn validate_remote_sink(param: &SinkParam) -> anyhow::Result<()> {
+async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> anyhow::Result<()> {
+    if sink_name == ElasticSearchSink::SINK_NAME
+        && param.downstream_pk.len() > 1
+        && param.properties.get(ES_OPTION_DELIMITER).is_none()
+    {
+        return Err(anyhow_error!(
+            "Es sink only support single pk or pk with delimiter option"
+        ));
+    }
     // FIXME: support struct and array in stream sink
     param.columns.iter().map(|col| {
-        if matches!(
-                col.data_type,
-                DataType::Int16
+        match &col.data_type {
+            DataType::Int16
                     | DataType::Int32
                     | DataType::Int64
                     | DataType::Float32
@@ -173,18 +183,34 @@ async fn validate_remote_sink(param: &SinkParam) -> anyhow::Result<()> {
                     | DataType::Time
                     | DataType::Interval
                     | DataType::Jsonb
-                    | DataType::Bytea
-                    | DataType::List(_)
-            ) {
-            Ok(())
-        } else {
-            Err(SinkError::Remote(anyhow_error!(
-                    "remote sink supports Int16, Int32, Int64, Float32, Float64, Boolean, Decimal, Time, Date, Interval, Jsonb, Timestamp, Timestamptz, List, Bytea and Varchar, got {:?}: {:?}",
-                    col.name,
-                    col.data_type,
-                )))
-        }
-    }).try_collect()?;
+                    | DataType::Bytea => Ok(()),
+            DataType::List(list) => {
+                if (sink_name==ElasticSearchSink::SINK_NAME) | matches!(list.as_ref(), DataType::Int16 | DataType::Int32 | DataType::Int64 | DataType::Float32 | DataType::Float64 | DataType::Varchar){
+                    Ok(())
+                } else{
+                    Err(SinkError::Remote(anyhow_error!(
+                        "Remote sink only support list<int16, int32, int64, float, double, varchar>, got {:?}: {:?}",
+                        col.name,
+                        col.data_type,
+                    )))
+                }
+            },
+            DataType::Struct(_) => {
+                if sink_name==ElasticSearchSink::SINK_NAME{
+                    Ok(())
+                }else{
+                    Err(SinkError::Remote(anyhow_error!(
+                        "Only Es sink support struct, got {:?}: {:?}",
+                        col.name,
+                        col.data_type,
+                    )))
+                }
+            },
+            DataType::Serial | DataType::Int256 => Err(SinkError::Remote(anyhow_error!(
+                            "remote sink supports Int16, Int32, Int64, Float32, Float64, Boolean, Decimal, Time, Date, Interval, Jsonb, Timestamp, Timestamptz, Bytea, List and Varchar, (Es sink support Struct) got {:?}: {:?}",
+                            col.name,
+                            col.data_type,
+                        )))}}).try_collect()?;
 
     let jvm = JVM.get_or_init()?;
     let sink_param = param.to_proto();
@@ -226,15 +252,34 @@ pub struct RemoteLogSinker {
     request_sender: BidiStreamSender<JniSinkWriterStreamRequest>,
     response_stream: BidiStreamReceiver<SinkWriterStreamResponse>,
     sink_metrics: SinkMetrics,
+    stream_chunk_converter: StreamChunkConverter,
 }
 
 impl RemoteLogSinker {
-    async fn new(sink_param: SinkParam, writer_param: SinkWriterParam) -> Result<Self> {
+    async fn new(
+        sink_param: SinkParam,
+        writer_param: SinkWriterParam,
+        sink_name: &str,
+    ) -> Result<Self> {
+        let sink_proto = sink_param.to_proto();
+        let payload_schema = if sink_name == ElasticSearchSink::SINK_NAME {
+            let columns = vec![
+                ColumnDesc::unnamed(ColumnId::from(0), DataType::Varchar).to_protobuf(),
+                ColumnDesc::unnamed(ColumnId::from(1), DataType::Jsonb).to_protobuf(),
+            ];
+            Some(TableSchema {
+                columns,
+                pk_indices: vec![],
+            })
+        } else {
+            sink_proto.table_schema.clone()
+        };
+
         let SinkWriterStreamHandle {
             request_sender,
             response_stream,
         } = EmbeddedConnectorClient::new()?
-            .start_sink_writer_stream(sink_param, SinkPayloadFormat::StreamChunk)
+            .start_sink_writer_stream(payload_schema, sink_proto, SinkPayloadFormat::StreamChunk)
             .await?;
 
         let sink_metrics = writer_param.sink_metrics;
@@ -242,6 +287,12 @@ impl RemoteLogSinker {
             request_sender,
             response_stream,
             sink_metrics,
+            stream_chunk_converter: StreamChunkConverter::new(
+                sink_name,
+                sink_param.schema(),
+                &sink_param.downstream_pk,
+                &sink_param.properties,
+            )?,
         })
     }
 }
@@ -385,6 +436,7 @@ impl LogSinker for RemoteLogSinker {
                                     .connector_sink_rows_received
                                     .inc_by(cardinality as _);
 
+                                let chunk = self.stream_chunk_converter.convert_chunk(chunk)?;
                                 request_tx
                                     .send_request(JniSinkWriterStreamRequest::Chunk {
                                         epoch,
@@ -451,7 +503,7 @@ impl<R: RemoteSinkTrait> Sink for CoordinatedRemoteSink<R> {
     const SINK_NAME: &'static str = R::SINK_NAME;
 
     async fn validate(&self) -> Result<()> {
-        validate_remote_sink(&self.param).await?;
+        validate_remote_sink(&self.param, Self::SINK_NAME).await?;
         Ok(())
     }
 
@@ -498,8 +550,13 @@ impl CoordinatedRemoteSinkWriter {
         connector_params: ConnectorParams,
         sink_metrics: SinkMetrics,
     ) -> Result<Self> {
+        let sink_proto = param.to_proto();
         let stream_handle = EmbeddedConnectorClient::new()?
-            .start_sink_writer_stream(param.clone(), connector_params.sink_payload_format)
+            .start_sink_writer_stream(
+                sink_proto.table_schema.clone(),
+                sink_proto,
+                connector_params.sink_payload_format,
+            )
             .await?;
 
         Ok(Self {
@@ -635,14 +692,16 @@ impl EmbeddedConnectorClient {
 
     async fn start_sink_writer_stream(
         &self,
-        sink_param: SinkParam,
+        payload_schema: Option<TableSchema>,
+        sink_proto: PbSinkParam,
         sink_payload_format: SinkPayloadFormat,
     ) -> Result<SinkWriterStreamHandle<JniSinkWriterStreamRequest>> {
         let (handle, first_rsp) = SinkWriterStreamHandle::initialize(
             SinkWriterStreamRequest {
                 request: Some(SinkRequest::Start(StartSink {
-                    sink_param: Some(sink_param.to_proto()),
+                    sink_param: Some(sink_proto),
                     format: sink_payload_format as i32,
+                    payload_schema,
                 })),
             },
             |rx| async move {
diff --git a/src/connector/src/source/google_pubsub/enumerator/client.rs b/src/connector/src/source/google_pubsub/enumerator/client.rs
index 03aec067cc511..e957bac8a641d 100644
--- a/src/connector/src/source/google_pubsub/enumerator/client.rs
+++ b/src/connector/src/source/google_pubsub/enumerator/client.rs
@@ -15,7 +15,7 @@
 use anyhow::{anyhow, bail};
 use async_trait::async_trait;
 use chrono::{TimeZone, Utc};
-use google_cloud_pubsub::client::Client;
+use google_cloud_pubsub::client::{Client, ClientConfig};
 use google_cloud_pubsub::subscription::{SeekTo, SubscriptionConfig};
 
 use crate::source::base::SplitEnumerator;
@@ -51,7 +51,8 @@ impl SplitEnumerator for PubsubSplitEnumerator {
         properties.initialize_env();
 
         // Validate config
-        let client = Client::new(Default::default())
+        let config = ClientConfig::default().with_auth().await?;
+        let client = Client::new(config)
             .await
             .map_err(|e| anyhow!("error initializing pubsub client: {:?}", e))?;
 
diff --git a/src/connector/src/source/google_pubsub/source/reader.rs b/src/connector/src/source/google_pubsub/source/reader.rs
index bbdbd934de83a..dfaf21ad3164e 100644
--- a/src/connector/src/source/google_pubsub/source/reader.rs
+++ b/src/connector/src/source/google_pubsub/source/reader.rs
@@ -16,7 +16,7 @@ use anyhow::{anyhow, ensure, Context, Result};
 use async_trait::async_trait;
 use chrono::{NaiveDateTime, TimeZone, Utc};
 use futures_async_stream::try_stream;
-use google_cloud_pubsub::client::Client;
+use google_cloud_pubsub::client::{Client, ClientConfig};
 use google_cloud_pubsub::subscription::{SeekTo, Subscription};
 use risingwave_common::bail;
 use tonic::Code;
@@ -126,9 +126,8 @@ impl SplitReader for PubsubSplitReader {
         // Set environment variables consumed by `google_cloud_pubsub`
         properties.initialize_env();
 
-        let client = Client::new(Default::default())
-            .await
-            .map_err(|e| anyhow!(e))?;
+        let config = ClientConfig::default().with_auth().await?;
+        let client = Client::new(config).await.map_err(|e| anyhow!(e))?;
         let subscription = client.subscription(&properties.subscription);
 
         if let Some(ref offset) = split.start_offset {
diff --git a/src/frontend/src/binder/expr/function.rs b/src/frontend/src/binder/expr/function.rs
index b96926478cda8..c8558b0756b5d 100644
--- a/src/frontend/src/binder/expr/function.rs
+++ b/src/frontend/src/binder/expr/function.rs
@@ -997,6 +997,7 @@ impl Binder {
                     }
                 ))),
                 ("pg_get_indexdef", raw_call(ExprType::PgGetIndexdef)),
+                ("pg_get_viewdef", raw_call(ExprType::PgGetViewdef)),
                 ("pg_relation_size", dispatch_by_len(vec![
                     (1, raw(|binder, inputs|{
                         let table_name = &inputs[0];
diff --git a/src/frontend/src/catalog/root_catalog.rs b/src/frontend/src/catalog/root_catalog.rs
index ec498cd60f625..bd411577476ba 100644
--- a/src/frontend/src/catalog/root_catalog.rs
+++ b/src/frontend/src/catalog/root_catalog.rs
@@ -579,6 +579,21 @@ impl Catalog {
             .ok_or_else(|| CatalogError::NotFound("table id", table_id.to_string()))
     }
 
+    /// This function is similar to `get_table_by_id` expect that a table must be in a given database.
+    pub fn get_table_by_id_with_db(
+        &self,
+        db_name: &str,
+        table_id: u32,
+    ) -> CatalogResult<&Arc<TableCatalog>> {
+        let table_id = TableId::from(table_id);
+        for schema in self.get_database_by_name(db_name)?.iter_schemas() {
+            if let Some(table) = schema.get_table_by_id(&table_id) {
+                return Ok(table);
+            }
+        }
+        Err(CatalogError::NotFound("table id", table_id.to_string()))
+    }
+
     pub fn get_schema_by_table_id(
         &self,
         db_name: &str,
@@ -714,6 +729,15 @@ impl Catalog {
             .ok_or_else(|| CatalogError::NotFound("view", view_name.to_string()))
     }
 
+    pub fn get_view_by_id(&self, db_name: &str, view_id: u32) -> CatalogResult<Arc<ViewCatalog>> {
+        for schema in self.get_database_by_name(db_name)?.iter_schemas() {
+            if let Some(view) = schema.get_view_by_id(&ViewId::from(view_id)) {
+                return Ok(view.clone());
+            }
+        }
+        Err(CatalogError::NotFound("view", view_id.to_string()))
+    }
+
     pub fn get_connection_by_name<'a>(
         &self,
         db_name: &str,
diff --git a/src/frontend/src/expr/function_impl/mod.rs b/src/frontend/src/expr/function_impl/mod.rs
index 2baf7f0efa7df..0541c392d1878 100644
--- a/src/frontend/src/expr/function_impl/mod.rs
+++ b/src/frontend/src/expr/function_impl/mod.rs
@@ -16,3 +16,4 @@ mod cast_regclass;
 mod col_description;
 pub mod context;
 mod pg_get_indexdef;
+mod pg_get_viewdef;
diff --git a/src/frontend/src/expr/function_impl/pg_get_viewdef.rs b/src/frontend/src/expr/function_impl/pg_get_viewdef.rs
new file mode 100644
index 0000000000000..d2e9caf9f6f29
--- /dev/null
+++ b/src/frontend/src/expr/function_impl/pg_get_viewdef.rs
@@ -0,0 +1,67 @@
+// Copyright 2024 RisingWave Labs
+//
+// 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.
+
+use std::fmt::Write;
+
+use anyhow::anyhow;
+use risingwave_expr::{capture_context, function, ExprError, Result};
+use risingwave_sqlparser::ast::Statement;
+use risingwave_sqlparser::parser::Parser;
+
+use super::context::{CATALOG_READER, DB_NAME};
+use crate::catalog::CatalogReader;
+
+#[function("pg_get_viewdef(int4) -> varchar")]
+fn pg_get_viewdef(oid: i32, writer: &mut impl Write) -> Result<()> {
+    pg_get_viewdef_impl_captured(oid, writer)
+}
+
+#[capture_context(CATALOG_READER, DB_NAME)]
+fn pg_get_viewdef_impl(
+    catalog: &CatalogReader,
+    db_name: &str,
+    oid: i32,
+    writer: &mut impl Write,
+) -> Result<()> {
+    let catalog_reader = catalog.read_guard();
+
+    if let Ok(view) = catalog_reader.get_view_by_id(db_name, oid as u32) {
+        write!(writer, "{}", view.sql).unwrap();
+        Ok(())
+    } else if let Ok(mv) = catalog_reader.get_table_by_id_with_db(db_name, oid as u32) {
+        let stmts = Parser::parse_sql(&mv.definition).map_err(|e| anyhow!(e))?;
+        let [stmt]: [_; 1] = stmts.try_into().unwrap();
+
+        if let Statement::CreateView {
+            query,
+            materialized,
+            ..
+        } = stmt
+            && materialized
+        {
+            write!(writer, "{}", query).unwrap();
+            Ok(())
+        } else {
+            Err(ExprError::InvalidParam {
+                name: "oid",
+                reason: format!("view or materialized view does not exist: {oid}").into(),
+            })
+        }
+    } else {
+        Err(ExprError::InvalidParam {
+            name: "oid",
+            reason: format!("view or materialized view does not exist: {oid}").into(),
+        })
+    }
+}
diff --git a/src/frontend/src/expr/pure.rs b/src/frontend/src/expr/pure.rs
index c22f19e3067a5..08b6bb47fa042 100644
--- a/src/frontend/src/expr/pure.rs
+++ b/src/frontend/src/expr/pure.rs
@@ -242,6 +242,7 @@ impl ExprVisitor for ImpureAnalyzer {
             | expr_node::Type::CastRegclass
             | expr_node::Type::PgGetIndexdef
             | expr_node::Type::ColDescription
+            | expr_node::Type::PgGetViewdef
             | expr_node::Type::MakeTimestamptz => self.impure = true,
         }
     }
diff --git a/src/frontend/src/handler/create_source.rs b/src/frontend/src/handler/create_source.rs
index a315fd88fc9bc..80e526b01c15f 100644
--- a/src/frontend/src/handler/create_source.rs
+++ b/src/frontend/src/handler/create_source.rs
@@ -53,8 +53,8 @@ use risingwave_pb::catalog::{
 use risingwave_pb::plan_common::{AdditionalColumnType, EncodeType, FormatType};
 use risingwave_pb::stream_plan::stream_fragment_graph::Parallelism;
 use risingwave_sqlparser::ast::{
-    get_delimiter, AstString, AvroSchema, ColumnDef, ConnectorSchema, CreateSourceStatement,
-    DebeziumAvroSchema, Encode, Format, Ident, ProtobufSchema, SourceWatermark,
+    get_delimiter, AstString, ColumnDef, ConnectorSchema, CreateSourceStatement, Encode, Format,
+    Ident, ProtobufSchema, SourceWatermark,
 };
 
 use super::RwPgResponse;
@@ -130,14 +130,20 @@ async fn extract_avro_table_schema(
     info: &StreamSourceInfo,
     with_properties: &HashMap<String, String>,
     format_encode_options: &mut BTreeMap<String, String>,
+    is_debezium: bool,
 ) -> Result<Vec<ColumnCatalog>> {
     let parser_config = SpecificParserConfig::new(info, with_properties)?;
     try_consume_string_from_options(format_encode_options, SCHEMA_REGISTRY_USERNAME);
     try_consume_string_from_options(format_encode_options, SCHEMA_REGISTRY_PASSWORD);
     consume_aws_config_from_options(format_encode_options);
 
-    let conf = AvroParserConfig::new(parser_config.encoding_config).await?;
-    let vec_column_desc = conf.map_to_columns()?;
+    let vec_column_desc = if is_debezium {
+        let conf = DebeziumAvroParserConfig::new(parser_config.encoding_config).await?;
+        conf.map_to_columns()?
+    } else {
+        let conf = AvroParserConfig::new(parser_config.encoding_config).await?;
+        conf.map_to_columns()?
+    };
     Ok(vec_column_desc
         .into_iter()
         .map(|col| ColumnCatalog {
@@ -156,29 +162,6 @@ async fn extract_debezium_avro_table_pk_columns(
     Ok(conf.extract_pks()?.drain(..).map(|c| c.name).collect())
 }
 
-// Map an Avro schema to a relational schema and return the pk_column_ids.
-async fn extract_debezium_avro_table_schema(
-    info: &StreamSourceInfo,
-    with_properties: &HashMap<String, String>,
-    format_encode_options: &mut BTreeMap<String, String>,
-) -> Result<Vec<ColumnCatalog>> {
-    let parser_config = SpecificParserConfig::new(info, with_properties)?;
-    try_consume_string_from_options(format_encode_options, SCHEMA_REGISTRY_USERNAME);
-    try_consume_string_from_options(format_encode_options, SCHEMA_REGISTRY_PASSWORD);
-    consume_aws_config_from_options(format_encode_options);
-
-    let conf = DebeziumAvroParserConfig::new(parser_config.encoding_config).await?;
-    let vec_column_desc = conf.map_to_columns()?;
-    let column_catalog = vec_column_desc
-        .into_iter()
-        .map(|col| ColumnCatalog {
-            column_desc: col.into(),
-            is_hidden: false,
-        })
-        .collect_vec();
-    Ok(column_catalog)
-}
-
 /// Map a protobuf schema to a relational schema.
 async fn extract_protobuf_table_schema(
     schema: &ProtobufSchema,
@@ -288,7 +271,6 @@ pub(crate) async fn bind_columns_from_source(
     session: &SessionImpl,
     source_schema: &ConnectorSchema,
     with_properties: &HashMap<String, String>,
-    create_cdc_source_job: bool,
 ) -> Result<(Option<Vec<ColumnCatalog>>, StreamSourceInfo)> {
     const MESSAGE_NAME_KEY: &str = "message";
     const KEY_MESSAGE_NAME_KEY: &str = "key.message";
@@ -298,14 +280,15 @@ pub(crate) async fn bind_columns_from_source(
     let format_encode_options = WithOptions::try_from(source_schema.row_options())?.into_inner();
     let mut format_encode_options_to_consume = format_encode_options.clone();
 
-    let get_key_message_name = |options: &mut BTreeMap<String, String>| -> Option<String> {
+    fn get_key_message_name(options: &mut BTreeMap<String, String>) -> Option<String> {
         consume_string_from_options(options, KEY_MESSAGE_NAME_KEY)
             .map(|ele| Some(ele.0))
             .unwrap_or(None)
-    };
-    let get_sr_name_strategy_check = |options: &mut BTreeMap<String, String>,
-                                      use_sr: bool|
-     -> Result<Option<i32>> {
+    }
+    fn get_sr_name_strategy_check(
+        options: &mut BTreeMap<String, String>,
+        use_sr: bool,
+    ) -> Result<Option<i32>> {
         let name_strategy = get_name_strategy_or_default(try_consume_string_from_options(
             options,
             NAME_STRATEGY_KEY,
@@ -316,18 +299,19 @@ pub(crate) async fn bind_columns_from_source(
             )));
         }
         Ok(name_strategy)
+    }
+
+    let mut stream_source_info = StreamSourceInfo {
+        format: format_to_prost(&source_schema.format) as i32,
+        row_encode: row_encode_to_prost(&source_schema.row_encode) as i32,
+        format_encode_options,
+        ..Default::default()
     };
 
-    let res = match (&source_schema.format, &source_schema.row_encode) {
-        (Format::Native, Encode::Native) => (
-            None,
-            StreamSourceInfo {
-                format: FormatType::Native as i32,
-                row_encode: EncodeType::Native as i32,
-                format_encode_options,
-                ..Default::default()
-            },
-        ),
+    let columns = match (&source_schema.format, &source_schema.row_encode) {
+        (Format::Native, Encode::Native)
+        | (Format::Plain, Encode::Bytes)
+        | (Format::DebeziumMongo, Encode::Json) => None,
         (Format::Plain, Encode::Protobuf) => {
             let (row_schema_location, use_schema_registry) =
                 get_schema_location(&mut format_encode_options_to_consume)?;
@@ -344,93 +328,58 @@ pub(crate) async fn bind_columns_from_source(
                 protobuf_schema.use_schema_registry,
             )?;
 
-            (
-                Some(
-                    extract_protobuf_table_schema(
-                        &protobuf_schema,
-                        with_properties,
-                        &mut format_encode_options_to_consume,
-                    )
-                    .await?,
-                ),
-                StreamSourceInfo {
-                    format: FormatType::Plain as i32,
-                    row_encode: EncodeType::Protobuf as i32,
-                    row_schema_location: protobuf_schema.row_schema_location.0.clone(),
-                    use_schema_registry: protobuf_schema.use_schema_registry,
-                    proto_message_name: protobuf_schema.message_name.0.clone(),
-                    key_message_name: get_key_message_name(&mut format_encode_options_to_consume),
-                    name_strategy: name_strategy
-                        .unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32),
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
-        }
-        (Format::Plain, Encode::Json) => {
-            let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
-            let columns = if create_cdc_source_job {
-                Some(debezium_cdc_source_schema())
-            } else {
-                extract_json_table_schema(
-                    &schema_config,
+            stream_source_info.use_schema_registry = protobuf_schema.use_schema_registry;
+            stream_source_info.row_schema_location = protobuf_schema.row_schema_location.0.clone();
+            stream_source_info.proto_message_name = protobuf_schema.message_name.0.clone();
+            stream_source_info.key_message_name =
+                get_key_message_name(&mut format_encode_options_to_consume);
+            stream_source_info.name_strategy =
+                name_strategy.unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32);
+
+            Some(
+                extract_protobuf_table_schema(
+                    &protobuf_schema,
                     with_properties,
                     &mut format_encode_options_to_consume,
                 )
-                .await?
-            };
-
-            (
-                columns,
-                StreamSourceInfo {
-                    format: FormatType::Plain as i32,
-                    row_encode: EncodeType::Json as i32,
-                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
-                    cdc_source_job: create_cdc_source_job,
-                    format_encode_options,
-                    ..Default::default()
-                },
+                .await?,
             )
         }
-        (Format::Plain, Encode::Avro) => {
+        (format @ (Format::Plain | Format::Upsert | Format::Debezium), Encode::Avro) => {
             let (row_schema_location, use_schema_registry) =
                 get_schema_location(&mut format_encode_options_to_consume)?;
-            let avro_schema = AvroSchema {
-                row_schema_location,
-                use_schema_registry,
-            };
 
-            let key_message_name = get_key_message_name(&mut format_encode_options_to_consume);
+            if matches!(format, Format::Debezium) && !use_schema_registry {
+                return Err(RwError::from(ProtocolError(
+                    "schema location for DEBEZIUM_AVRO row format is not supported".to_string(),
+                )));
+            }
+
             let message_name = try_consume_string_from_options(
                 &mut format_encode_options_to_consume,
                 MESSAGE_NAME_KEY,
             );
             let name_strategy = get_sr_name_strategy_check(
                 &mut format_encode_options_to_consume,
-                avro_schema.use_schema_registry,
+                use_schema_registry,
             )?;
-            let stream_source_info = StreamSourceInfo {
-                format: FormatType::Plain as i32,
-                row_encode: EncodeType::Avro as i32,
-                row_schema_location: avro_schema.row_schema_location.0.clone(),
-                use_schema_registry: avro_schema.use_schema_registry,
-                proto_message_name: message_name.unwrap_or(AstString("".into())).0,
-                key_message_name,
-                name_strategy: name_strategy
-                    .unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32),
-                format_encode_options,
-                ..Default::default()
-            };
-            (
-                Some(
-                    extract_avro_table_schema(
-                        &stream_source_info,
-                        with_properties,
-                        &mut format_encode_options_to_consume,
-                    )
-                    .await?,
-                ),
-                stream_source_info,
+
+            stream_source_info.use_schema_registry = use_schema_registry;
+            stream_source_info.row_schema_location = row_schema_location.0.clone();
+            stream_source_info.proto_message_name = message_name.unwrap_or(AstString("".into())).0;
+            stream_source_info.key_message_name =
+                get_key_message_name(&mut format_encode_options_to_consume);
+            stream_source_info.name_strategy =
+                name_strategy.unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32);
+
+            Some(
+                extract_avro_table_schema(
+                    &stream_source_info,
+                    with_properties,
+                    &mut format_encode_options_to_consume,
+                    matches!(format, Format::Debezium),
+                )
+                .await?,
             )
         }
         (Format::Plain, Encode::Csv) => {
@@ -451,204 +400,81 @@ pub(crate) async fn bind_columns_from_source(
                         .to_owned(),
                 )));
             }
-            (
-                None,
-                StreamSourceInfo {
-                    format: FormatType::Plain as i32,
-                    row_encode: EncodeType::Csv as i32,
-                    csv_delimiter: delimiter as i32,
-                    csv_has_header: has_header,
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
-        }
-        (Format::Plain, Encode::Bytes) => (
-            None,
-            StreamSourceInfo {
-                format: FormatType::Plain as i32,
-                row_encode: EncodeType::Bytes as i32,
-                format_encode_options,
-                ..Default::default()
-            },
-        ),
-        (Format::Upsert, Encode::Json) => {
-            let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
-            let columns = extract_json_table_schema(
-                &schema_config,
-                with_properties,
-                &mut format_encode_options_to_consume,
-            )
-            .await?;
 
-            (
-                columns,
-                StreamSourceInfo {
-                    format: FormatType::Upsert as i32,
-                    row_encode: EncodeType::Json as i32,
-                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
-        }
-        (Format::Upsert, Encode::Avro) => {
-            let (row_schema_location, use_schema_registry) =
-                get_schema_location(&mut format_encode_options_to_consume)?;
-            let avro_schema = AvroSchema {
-                row_schema_location,
-                use_schema_registry,
-            };
-
-            let name_strategy = get_sr_name_strategy_check(
-                &mut format_encode_options_to_consume,
-                avro_schema.use_schema_registry,
-            )?
-            .unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32);
-            let key_message_name = get_key_message_name(&mut format_encode_options_to_consume);
-            let message_name = try_consume_string_from_options(
-                &mut format_encode_options_to_consume,
-                MESSAGE_NAME_KEY,
-            );
+            stream_source_info.csv_delimiter = delimiter as i32;
+            stream_source_info.csv_has_header = has_header;
 
-            let stream_source_info = StreamSourceInfo {
-                key_message_name,
-                format: FormatType::Upsert as i32,
-                row_encode: EncodeType::Avro as i32,
-                row_schema_location: avro_schema.row_schema_location.0.clone(),
-                use_schema_registry: avro_schema.use_schema_registry,
-                proto_message_name: message_name.unwrap_or(AstString("".into())).0,
-                name_strategy,
-                format_encode_options,
-                ..Default::default()
-            };
-            let columns = extract_avro_table_schema(
-                &stream_source_info,
-                with_properties,
-                &mut format_encode_options_to_consume,
-            )
-            .await?;
-
-            (Some(columns), stream_source_info)
+            None
         }
-
-        (Format::Debezium, Encode::Json) => {
+        (
+            Format::Plain | Format::Upsert | Format::Maxwell | Format::Canal | Format::Debezium,
+            Encode::Json,
+        ) => {
             let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
-            (
-                extract_json_table_schema(
-                    &schema_config,
-                    with_properties,
-                    &mut format_encode_options_to_consume,
-                )
-                .await?,
-                StreamSourceInfo {
-                    format: FormatType::Debezium as i32,
-                    row_encode: EncodeType::Json as i32,
-                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
-        }
-        (Format::Debezium, Encode::Avro) => {
-            let (row_schema_location, use_schema_registry) =
-                get_schema_location(&mut format_encode_options_to_consume)?;
-            if !use_schema_registry {
-                return Err(RwError::from(ProtocolError(
-                    "schema location for DEBEZIUM_AVRO row format is not supported".to_string(),
-                )));
-            }
-            let avro_schema = DebeziumAvroSchema {
-                row_schema_location,
-            };
-
-            // no need to check whether works schema registry because debezium avro always work with
-            // schema registry
-            let name_strategy =
-                get_sr_name_strategy_check(&mut format_encode_options_to_consume, true)?;
-            let message_name = try_consume_string_from_options(
-                &mut format_encode_options_to_consume,
-                MESSAGE_NAME_KEY,
-            );
-            let key_message_name = get_key_message_name(&mut format_encode_options_to_consume);
-
-            let stream_source_info = StreamSourceInfo {
-                use_schema_registry,
-                proto_message_name: message_name.unwrap_or(AstString("".into())).0,
-                name_strategy: name_strategy
-                    .unwrap_or(PbSchemaRegistryNameStrategy::Unspecified as i32),
-                format: FormatType::Debezium as i32,
-                row_encode: EncodeType::Avro as i32,
-                row_schema_location: avro_schema.row_schema_location.0.clone(),
-                key_message_name,
-                format_encode_options,
-                ..Default::default()
-            };
+            stream_source_info.use_schema_registry =
+                json_schema_infer_use_schema_registry(&schema_config);
 
-            let full_columns = extract_debezium_avro_table_schema(
-                &stream_source_info,
+            extract_json_table_schema(
+                &schema_config,
                 with_properties,
                 &mut format_encode_options_to_consume,
             )
-            .await?;
-
-            (Some(full_columns), stream_source_info)
+            .await?
         }
-        (Format::DebeziumMongo, Encode::Json) => (
-            None,
-            StreamSourceInfo {
-                format: FormatType::DebeziumMongo as i32,
-                row_encode: EncodeType::Json as i32,
-                format_encode_options,
-                ..Default::default()
-            },
-        ),
-
-        (Format::Maxwell, Encode::Json) => {
-            let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
-            (
-                extract_json_table_schema(
-                    &schema_config,
-                    with_properties,
-                    &mut format_encode_options_to_consume,
-                )
-                .await?,
-                StreamSourceInfo {
-                    format: FormatType::Maxwell as i32,
-                    row_encode: EncodeType::Json as i32,
-                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
+        (format, encoding) => {
+            return Err(RwError::from(ProtocolError(format!(
+                "Unknown combination {:?} {:?}",
+                format, encoding
+            ))));
         }
+    };
 
-        (Format::Canal, Encode::Json) => {
-            let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
-            (
-                extract_json_table_schema(
-                    &schema_config,
-                    with_properties,
-                    &mut format_encode_options_to_consume,
-                )
-                .await?,
-                StreamSourceInfo {
-                    format: FormatType::Canal as i32,
-                    row_encode: EncodeType::Json as i32,
-                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
-                    format_encode_options,
-                    ..Default::default()
-                },
-            )
-        }
+    if !format_encode_options_to_consume.is_empty() {
+        let err_string = format!(
+            "Get unknown format_encode_options for {:?} {:?}: {}",
+            source_schema.format,
+            source_schema.row_encode,
+            format_encode_options_to_consume
+                .keys()
+                .map(|k| k.to_string())
+                .collect::<Vec<String>>()
+                .join(","),
+        );
+        session.notice_to_user(err_string);
+    }
+    Ok((columns, stream_source_info))
+}
+
+fn bind_columns_from_source_for_cdc(
+    session: &SessionImpl,
+    source_schema: &ConnectorSchema,
+    _with_properties: &HashMap<String, String>,
+) -> Result<(Option<Vec<ColumnCatalog>>, StreamSourceInfo)> {
+    let format_encode_options = WithOptions::try_from(source_schema.row_options())?.into_inner();
+    let mut format_encode_options_to_consume = format_encode_options.clone();
+
+    match (&source_schema.format, &source_schema.row_encode) {
+        (Format::Plain, Encode::Json) => (),
         (format, encoding) => {
+            // Note: parser will also check this. Just be extra safe here
             return Err(RwError::from(ProtocolError(format!(
-                "Unknown combination {:?} {:?}",
+                "Row format for CDC connectors should be either omitted or set to `FORMAT PLAIN ENCODE JSON`, got: {:?} {:?}",
                 format, encoding
             ))));
         }
     };
 
+    let columns = debezium_cdc_source_schema();
+    let schema_config = get_json_schema_location(&mut format_encode_options_to_consume)?;
+
+    let stream_source_info = StreamSourceInfo {
+        format: format_to_prost(&source_schema.format) as i32,
+        row_encode: row_encode_to_prost(&source_schema.row_encode) as i32,
+        format_encode_options,
+        use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
+        cdc_source_job: true,
+        ..Default::default()
+    };
     if !format_encode_options_to_consume.is_empty() {
         let err_string = format!(
             "Get unknown format_encode_options for {:?} {:?}: {}",
@@ -662,7 +488,7 @@ pub(crate) async fn bind_columns_from_source(
         );
         session.notice_to_user(err_string);
     }
-    Ok(res)
+    Ok((Some(columns), stream_source_info))
 }
 
 /// add connector-spec columns to the end of column catalog
@@ -1298,13 +1124,11 @@ pub async fn handle_create_source(
         false
     };
 
-    let (columns_from_resolve_source, source_info) = bind_columns_from_source(
-        &session,
-        &source_schema,
-        &with_properties,
-        create_cdc_source_job,
-    )
-    .await?;
+    let (columns_from_resolve_source, source_info) = if create_cdc_source_job {
+        bind_columns_from_source_for_cdc(&session, &source_schema, &with_properties)?
+    } else {
+        bind_columns_from_source(&session, &source_schema, &with_properties).await?
+    };
     let columns_from_sql = bind_sql_columns(&stmt.columns)?;
 
     let mut columns = bind_all_columns(
@@ -1434,6 +1258,29 @@ pub async fn handle_create_source(
     Ok(PgResponse::empty_result(StatementType::CREATE_SOURCE))
 }
 
+fn format_to_prost(format: &Format) -> FormatType {
+    match format {
+        Format::Native => FormatType::Native,
+        Format::Plain => FormatType::Plain,
+        Format::Upsert => FormatType::Upsert,
+        Format::Debezium => FormatType::Debezium,
+        Format::DebeziumMongo => FormatType::DebeziumMongo,
+        Format::Maxwell => FormatType::Maxwell,
+        Format::Canal => FormatType::Canal,
+    }
+}
+fn row_encode_to_prost(row_encode: &Encode) -> EncodeType {
+    match row_encode {
+        Encode::Native => EncodeType::Native,
+        Encode::Json => EncodeType::Json,
+        Encode::Avro => EncodeType::Avro,
+        Encode::Protobuf => EncodeType::Protobuf,
+        Encode::Csv => EncodeType::Csv,
+        Encode::Bytes => EncodeType::Bytes,
+        Encode::Template => EncodeType::Template,
+    }
+}
+
 #[cfg(test)]
 pub mod tests {
     use std::collections::HashMap;
diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
index d9df554d20320..02f0321932b79 100644
--- a/src/frontend/src/handler/create_table.rs
+++ b/src/frontend/src/handler/create_table.rs
@@ -478,13 +478,8 @@ pub(crate) async fn gen_create_table_plan_with_source(
 
     let sql_pk_names = bind_sql_pk_names(&column_defs, &constraints)?;
 
-    let (columns_from_resolve_source, source_info) = bind_columns_from_source(
-        context.session_ctx(),
-        &source_schema,
-        &with_properties,
-        false,
-    )
-    .await?;
+    let (columns_from_resolve_source, source_info) =
+        bind_columns_from_source(context.session_ctx(), &source_schema, &with_properties).await?;
     let columns_from_sql = bind_sql_columns(&column_defs)?;
 
     let mut columns = bind_all_columns(
diff --git a/src/meta/src/manager/catalog/fragment.rs b/src/meta/src/manager/catalog/fragment.rs
index e9e3fc7a66d3a..a4128aaeb8f95 100644
--- a/src/meta/src/manager/catalog/fragment.rs
+++ b/src/meta/src/manager/catalog/fragment.rs
@@ -616,14 +616,16 @@ impl FragmentManager {
                 if table_ids.contains(&dependent_table_id) {
                     continue;
                 }
-                let mut dependent_table = table_fragments
-                    .get_mut(dependent_table_id)
-                    .with_context(|| {
-                        format!(
+                let mut dependent_table =
+                    if let Some(dependent_table) = table_fragments.get_mut(dependent_table_id) {
+                        dependent_table
+                    } else {
+                        tracing::error!(
                             "dependent table_fragment not exist: id={}",
                             dependent_table_id
-                        )
-                    })?;
+                        );
+                        continue;
+                    };
 
                 dependent_table
                     .fragments
diff --git a/src/rpc_client/src/connector_client.rs b/src/rpc_client/src/connector_client.rs
index 85d9b6c9f6ec1..cd53cd019ea64 100644
--- a/src/rpc_client/src/connector_client.rs
+++ b/src/rpc_client/src/connector_client.rs
@@ -265,14 +265,16 @@ impl ConnectorClient {
 
     pub async fn start_sink_writer_stream(
         &self,
-        sink_param: SinkParam,
+        payload_schema: Option<TableSchema>,
+        sink_proto: PbSinkParam,
         sink_payload_format: SinkPayloadFormat,
     ) -> Result<SinkWriterStreamHandle> {
         let mut rpc_client = self.rpc_client.clone();
         let (handle, first_rsp) = SinkWriterStreamHandle::initialize(
             SinkWriterStreamRequest {
                 request: Some(SinkRequest::Start(StartSink {
-                    sink_param: Some(sink_param),
+                    payload_schema,
+                    sink_param: Some(sink_proto),
                     format: sink_payload_format as i32,
                 })),
             },
diff --git a/src/stream/src/executor/mview/materialize.rs b/src/stream/src/executor/mview/materialize.rs
index aa4262ae78f5a..ddc985528d7f4 100644
--- a/src/stream/src/executor/mview/materialize.rs
+++ b/src/stream/src/executor/mview/materialize.rs
@@ -19,14 +19,12 @@ use std::ops::Deref;
 use std::sync::Arc;
 
 use bytes::Bytes;
-use enum_as_inner::EnumAsInner;
 use futures::{stream, StreamExt};
 use futures_async_stream::try_stream;
 use itertools::Itertools;
 use risingwave_common::array::{Op, StreamChunk};
 use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::{ColumnDesc, ColumnId, ConflictBehavior, Schema, TableId};
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::row::{CompactedRow, RowDeserializer};
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
@@ -171,11 +169,7 @@ impl<S: StateStore, SD: ValueRowSerde> MaterializeExecutor<S, SD> {
 
                             let fixed_changes = self
                                 .materialize_cache
-                                .handle_conflict(
-                                    row_ops,
-                                    &self.state_table,
-                                    &self.conflict_behavior,
-                                )
+                                .handle(row_ops, &self.state_table, &self.conflict_behavior)
                                 .await?;
 
                             match generate_output(fixed_changes, data_types.clone())? {
@@ -337,7 +331,7 @@ impl MaterializeBuffer {
     }
 
     pub fn delete(&mut self, pk: Vec<u8>, old_value: Bytes) {
-        let entry = self.buffer.entry(pk);
+        let entry: Entry<'_, Vec<u8>, KeyOp> = self.buffer.entry(pk);
         match entry {
             Entry::Vacant(e) => {
                 e.insert(KeyOp::Delete(old_value));
@@ -419,13 +413,7 @@ pub struct MaterializeCache<SD> {
     _serde: PhantomData<SD>,
 }
 
-#[derive(EnumAsInner, EstimateSize)]
-pub enum CacheValue {
-    Overwrite(Option<CompactedRow>),
-    Ignore(Option<EmptyValue>),
-}
-
-type EmptyValue = ();
+type CacheValue = Option<CompactedRow>;
 
 impl<SD: ValueRowSerde> MaterializeCache<SD> {
     pub fn new(watermark_epoch: AtomicU64Ref, metrics_info: MetricsInfo) -> Self {
@@ -438,7 +426,7 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
         }
     }
 
-    pub async fn handle_conflict<'a, S: StateStore>(
+    pub async fn handle<'a, S: StateStore>(
         &mut self,
         row_ops: Vec<(Op, Vec<u8>, Bytes)>,
         table: &StateTableInner<S, SD>,
@@ -459,7 +447,7 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
                 Op::Insert | Op::UpdateInsert => {
                     match conflict_behavior {
                         ConflictBehavior::Overwrite => {
-                            match self.force_get(&key).as_overwrite().unwrap() {
+                            match self.force_get(&key) {
                                 Some(old_row) => fixed_changes.update(
                                     key.clone(),
                                     old_row.row.clone(),
@@ -470,7 +458,7 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
                             update_cache = true;
                         }
                         ConflictBehavior::IgnoreConflict => {
-                            match self.force_get(&key).as_ignore().unwrap() {
+                            match self.force_get(&key) {
                                 Some(_) => (),
                                 None => {
                                     fixed_changes.insert(key.clone(), value.clone());
@@ -484,13 +472,10 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
                     if update_cache {
                         match conflict_behavior {
                             ConflictBehavior::Overwrite => {
-                                self.data.push(
-                                    key,
-                                    CacheValue::Overwrite(Some(CompactedRow { row: value })),
-                                );
+                                self.data.push(key, Some(CompactedRow { row: value }));
                             }
                             ConflictBehavior::IgnoreConflict => {
-                                self.data.push(key, CacheValue::Ignore(Some(())));
+                                self.data.push(key, Some(CompactedRow { row: value }));
                             }
                             _ => unreachable!(),
                         }
@@ -500,7 +485,7 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
                 Op::Delete | Op::UpdateDelete => {
                     match conflict_behavior {
                         ConflictBehavior::Overwrite => {
-                            match self.force_get(&key).as_overwrite().unwrap() {
+                            match self.force_get(&key) {
                                 Some(old_row) => {
                                     fixed_changes.delete(key.clone(), old_row.row.clone());
                                 }
@@ -508,20 +493,22 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
                             };
                             update_cache = true;
                         }
-                        ConflictBehavior::IgnoreConflict => (),
+                        ConflictBehavior::IgnoreConflict => {
+                            match self.force_get(&key) {
+                                Some(old_row) => {
+                                    if old_row.row == value {
+                                        fixed_changes.delete(key.clone(), old_row.row.clone());
+                                        update_cache = true;
+                                    }
+                                }
+                                None => (), // delete a nonexistent value
+                            };
+                        }
                         _ => unreachable!(),
                     };
 
                     if update_cache {
-                        match conflict_behavior {
-                            ConflictBehavior::Overwrite => {
-                                self.data.push(key, CacheValue::Overwrite(None));
-                            }
-                            ConflictBehavior::IgnoreConflict => {
-                                self.data.push(key, CacheValue::Ignore(Some(())));
-                            }
-                            _ => unreachable!(),
-                        }
+                        self.data.push(key, None);
                     }
                 }
             }
@@ -560,10 +547,8 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
         while let Some(result) = buffered.next().await {
             let (key, value) = result;
             match conflict_behavior {
-                ConflictBehavior::Overwrite => self.data.push(key, CacheValue::Overwrite(value?)),
-                ConflictBehavior::IgnoreConflict => {
-                    self.data.push(key, CacheValue::Ignore(value?.map(|_| ())))
-                }
+                ConflictBehavior::Overwrite => self.data.push(key, value?),
+                ConflictBehavior::IgnoreConflict => self.data.push(key, value?),
                 _ => unreachable!(),
             };
         }
@@ -1245,6 +1230,104 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_ignore_delete_then_insert() {
+        // Prepare storage and memtable.
+        let memory_state_store = MemoryStateStore::new();
+        let table_id = TableId::new(1);
+        // Two columns of int32 type, the first column is PK.
+        let schema = Schema::new(vec![
+            Field::unnamed(DataType::Int32),
+            Field::unnamed(DataType::Int32),
+        ]);
+        let column_ids = vec![0.into(), 1.into()];
+
+        // test insert after delete one pk, the latter insert should succeed.
+        let chunk1 = StreamChunk::from_pretty(
+            " i i
+            + 1 3
+            - 1 3
+            + 1 6",
+        );
+
+        // Prepare stream executors.
+        let source = MockSource::with_messages(
+            schema.clone(),
+            PkIndices::new(),
+            vec![
+                Message::Barrier(Barrier::new_test_barrier(1)),
+                Message::Chunk(chunk1),
+                Message::Barrier(Barrier::new_test_barrier(2)),
+            ],
+        );
+
+        let order_types = vec![OrderType::ascending()];
+        let column_descs = vec![
+            ColumnDesc::unnamed(column_ids[0], DataType::Int32),
+            ColumnDesc::unnamed(column_ids[1], DataType::Int32),
+        ];
+
+        let table = StorageTable::for_test(
+            memory_state_store.clone(),
+            table_id,
+            column_descs,
+            order_types,
+            vec![0],
+            vec![0, 1],
+        );
+
+        let mut materialize_executor = Box::new(
+            MaterializeExecutor::for_test(
+                Box::new(source),
+                memory_state_store,
+                table_id,
+                vec![ColumnOrder::new(0, OrderType::ascending())],
+                column_ids,
+                1,
+                Arc::new(AtomicU64::new(0)),
+                ConflictBehavior::IgnoreConflict,
+            )
+            .await,
+        )
+        .execute();
+        let _msg1 = materialize_executor
+            .next()
+            .await
+            .transpose()
+            .unwrap()
+            .unwrap()
+            .as_barrier()
+            .unwrap();
+        let _msg2 = materialize_executor
+            .next()
+            .await
+            .transpose()
+            .unwrap()
+            .unwrap()
+            .as_chunk()
+            .unwrap();
+        let _msg3 = materialize_executor
+            .next()
+            .await
+            .transpose()
+            .unwrap()
+            .unwrap()
+            .as_barrier()
+            .unwrap();
+
+        let row = table
+            .get_row(
+                &OwnedRow::new(vec![Some(1_i32.into())]),
+                HummockReadEpoch::NoWait(u64::MAX),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            row,
+            Some(OwnedRow::new(vec![Some(1_i32.into()), Some(6_i32.into())]))
+        );
+    }
+
     #[tokio::test]
     async fn test_ignore_delete_and_update_conflict() {
         // Prepare storage and memtable.
diff --git a/src/stream/src/task/env.rs b/src/stream/src/task/env.rs
index 63e610ad69ef4..0083e8e02ff28 100644
--- a/src/stream/src/task/env.rs
+++ b/src/stream/src/task/env.rs
@@ -20,8 +20,6 @@ use risingwave_common::system_param::local_manager::LocalSystemParamsManagerRef;
 use risingwave_common::util::addr::HostAddr;
 use risingwave_connector::source::monitor::SourceMetrics;
 use risingwave_connector::ConnectorParams;
-#[cfg(test)]
-use risingwave_pb::connector_service::SinkPayloadFormat;
 use risingwave_rpc_client::MetaClient;
 use risingwave_source::dml_manager::DmlManagerRef;
 use risingwave_storage::StateStoreImpl;
@@ -94,6 +92,7 @@ impl StreamEnvironment {
     #[cfg(test)]
     pub fn for_test() -> Self {
         use risingwave_common::system_param::local_manager::LocalSystemParamsManager;
+        use risingwave_pb::connector_service::SinkPayloadFormat;
         use risingwave_source::dml_manager::DmlManager;
         use risingwave_storage::monitor::MonitoredStorageMetrics;
         StreamEnvironment {