From 0c9c0d03646575cb1b420b4045e007397d5e7ce3 Mon Sep 17 00:00:00 2001
From: Qian Xia <lauraxiaqian@gmail.com>
Date: Fri, 6 Dec 2024 19:32:33 +0800
Subject: [PATCH] [#5447][#5720][#5727] create topic, disable create table for
 hudi, enable schema properties and comment for partial paimon (#5771)

### What changes were proposed in this pull request?

1. Support creating topic/editing topic/deleting topic/viewing topic
<img width="1417" alt="image"
src="https://github.com/user-attachments/assets/24e0ae59-7ec5-4115-becc-2d8a83dcd8d6">

2. Fix issue with creating a table for hudi
<img width="1481" alt="image"
src="https://github.com/user-attachments/assets/bc64eaa9-1b8e-48b2-98b4-899ebfd6eb02">

3. [Improvement] Paimon catalog, schema properties can be set when
backend is jdbc and hive
<img width="1441" alt="image"
src="https://github.com/user-attachments/assets/9bb2746f-d3b0-4f05-84bc-d5baadb8999d">
<img width="1434" alt="image"
src="https://github.com/user-attachments/assets/537ec2fc-a4fa-48ff-bbd1-f606751a2a46">

### Why are the changes needed?
N/A

Fix: #5447, #5720, #5727

### Does this PR introduce _any_ user-facing change?
N/A

### How was this patch tested?
manually

---------

Co-authored-by: Qiming Teng <tengqm@outlook.com>
---
 .../src/app/metalakes/CreateMetalakeDialog.js |   4 +-
 .../rightContent/CreateCatalogDialog.js       |   6 +-
 .../rightContent/CreateFilesetDialog.js       |   6 +-
 .../rightContent/CreateSchemaDialog.js        |  40 +-
 .../rightContent/CreateTableDialog.js         |   6 +-
 .../rightContent/CreateTopicDialog.js         | 435 ++++++++++++++++++
 .../metalake/rightContent/RightContent.js     |  50 +-
 .../tabsContent/tableView/TableView.js        |  58 ++-
 web/web/src/lib/api/topics/index.js           |  20 +-
 web/web/src/lib/store/metalakes/index.js      | 108 ++++-
 10 files changed, 687 insertions(+), 46 deletions(-)
 create mode 100644 web/web/src/app/metalakes/metalake/rightContent/CreateTopicDialog.js

diff --git a/web/web/src/app/metalakes/CreateMetalakeDialog.js b/web/web/src/app/metalakes/CreateMetalakeDialog.js
index 8b8daaadb05..2fb70ff3823 100644
--- a/web/web/src/app/metalakes/CreateMetalakeDialog.js
+++ b/web/web/src/app/metalakes/CreateMetalakeDialog.js
@@ -316,8 +316,8 @@ const CreateMetalakeDialog = props => {
                       )}
                       {item.invalid && (
                         <FormHelperText className={'twc-text-error-main'}>
-                          Invalid key, matches strings starting with a letter/underscore, followed by alphanumeric
-                          characters, underscores, hyphens, or dots.
+                          Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                          underscores, hyphens, or dots.
                         </FormHelperText>
                       )}
                     </FormControl>
diff --git a/web/web/src/app/metalakes/metalake/rightContent/CreateCatalogDialog.js b/web/web/src/app/metalakes/metalake/rightContent/CreateCatalogDialog.js
index a4cf85623fb..cd9c101f762 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/CreateCatalogDialog.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/CreateCatalogDialog.js
@@ -680,12 +680,12 @@ const CreateCatalogDialog = props => {
                           )}
                           {item.key && item.invalid && (
                             <FormHelperText className={'twc-text-error-main'}>
-                              Invalid key, matches strings starting with a letter/underscore, followed by alphanumeric
-                              characters, underscores, hyphens, or dots.
+                              Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                              underscores, hyphens, or dots.
                             </FormHelperText>
                           )}
                           {!item.key.trim() && (
-                            <FormHelperText className={'twc-text-error-main'}>Key is required field</FormHelperText>
+                            <FormHelperText className={'twc-text-error-main'}>Key is required</FormHelperText>
                           )}
                         </FormControl>
                       </Grid>
diff --git a/web/web/src/app/metalakes/metalake/rightContent/CreateFilesetDialog.js b/web/web/src/app/metalakes/metalake/rightContent/CreateFilesetDialog.js
index cb19015458b..6a69d82f879 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/CreateFilesetDialog.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/CreateFilesetDialog.js
@@ -461,12 +461,12 @@ const CreateFilesetDialog = props => {
                         )}
                         {item.key && item.invalid && (
                           <FormHelperText className={'twc-text-error-main'}>
-                            Invalid key, matches strings starting with a letter/underscore, followed by alphanumeric
-                            characters, underscores, hyphens, or dots.
+                            Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                            underscores, hyphens, or dots.
                           </FormHelperText>
                         )}
                         {!item.key.trim() && (
-                          <FormHelperText className={'twc-text-error-main'}>Key is required field</FormHelperText>
+                          <FormHelperText className={'twc-text-error-main'}>Key is required</FormHelperText>
                         )}
                       </FormControl>
                     </Grid>
diff --git a/web/web/src/app/metalakes/metalake/rightContent/CreateSchemaDialog.js b/web/web/src/app/metalakes/metalake/rightContent/CreateSchemaDialog.js
index 08efd4036db..ce893802a19 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/CreateSchemaDialog.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/CreateSchemaDialog.js
@@ -23,20 +23,18 @@ import { useState, forwardRef, useEffect, Fragment } from 'react'
 
 import {
   Box,
-  Grid,
   Button,
   Dialog,
-  TextField,
-  Typography,
-  DialogContent,
   DialogActions,
-  IconButton,
+  DialogContent,
   Fade,
-  Select,
-  MenuItem,
-  InputLabel,
   FormControl,
-  FormHelperText
+  FormHelperText,
+  Grid,
+  IconButton,
+  InputLabel,
+  TextField,
+  Typography
 } from '@mui/material'
 
 import Icon from '@/components/Icon'
@@ -90,6 +88,10 @@ const CreateSchemaDialog = props => {
   const activatedCatalogDetail = store.activatedDetails
   const [cacheData, setCacheData] = useState()
 
+  const paimonCatalogBackend =
+    activatedCatalogDetail?.provider === 'lakehouse-paimon' &&
+    ['hive', 'jdbc'].includes(activatedCatalogDetail?.properties['catalog-backend'])
+
   const {
     control,
     reset,
@@ -293,7 +295,8 @@ const CreateSchemaDialog = props => {
               </FormControl>
             </Grid>
 
-            {!['jdbc-mysql', 'lakehouse-paimon', 'jdbc-oceanbase'].includes(activatedCatalogDetail?.provider) && (
+            {(!['jdbc-mysql', 'lakehouse-paimon', 'jdbc-oceanbase'].includes(activatedCatalogDetail?.provider) ||
+              paimonCatalogBackend) && (
               <Grid item xs={12}>
                 <FormControl fullWidth>
                   <Controller
@@ -315,12 +318,14 @@ const CreateSchemaDialog = props => {
                     )}
                   />
                 </FormControl>
+                {activatedCatalogDetail?.properties['catalog-backend']}
               </Grid>
             )}
 
-            {!['jdbc-postgresql', 'lakehouse-paimon', 'kafka', 'jdbc-mysql', 'jdbc-oceanbase'].includes(
+            {(!['jdbc-postgresql', 'lakehouse-paimon', 'kafka', 'jdbc-mysql', 'jdbc-oceanbase'].includes(
               activatedCatalogDetail?.provider
-            ) && (
+            ) ||
+              paimonCatalogBackend) && (
               <Grid item xs={12} data-refer='schema-props-layout'>
                 <Typography sx={{ mb: 2 }} variant='body2'>
                   Properties
@@ -385,12 +390,12 @@ const CreateSchemaDialog = props => {
                           )}
                           {item.key && item.invalid && (
                             <FormHelperText className={'twc-text-error-main'}>
-                              Invalid key, matches strings starting with a letter/underscore, followed by alphanumeric
-                              characters, underscores, hyphens, or dots.
+                              Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                              underscores, hyphens, or dots.
                             </FormHelperText>
                           )}
                           {!item.key.trim() && (
-                            <FormHelperText className={'twc-text-error-main'}>Key is required field</FormHelperText>
+                            <FormHelperText className={'twc-text-error-main'}>Key is required</FormHelperText>
                           )}
                         </FormControl>
                       </Grid>
@@ -400,9 +405,10 @@ const CreateSchemaDialog = props => {
               </Grid>
             )}
 
-            {!['jdbc-postgresql', 'lakehouse-paimon', 'kafka', 'jdbc-mysql', 'jdbc-oceanbase'].includes(
+            {(!['jdbc-postgresql', 'lakehouse-paimon', 'kafka', 'jdbc-mysql', 'jdbc-oceanbase'].includes(
               activatedCatalogDetail?.provider
-            ) && (
+            ) ||
+              paimonCatalogBackend) && (
               <Grid item xs={12}>
                 <Button
                   size='small'
diff --git a/web/web/src/app/metalakes/metalake/rightContent/CreateTableDialog.js b/web/web/src/app/metalakes/metalake/rightContent/CreateTableDialog.js
index b1a8a06f7ec..a1c337725e2 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/CreateTableDialog.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/CreateTableDialog.js
@@ -784,12 +784,12 @@ const CreateTableDialog = props => {
                         )}
                         {item.key && item.invalid && (
                           <FormHelperText className={'twc-text-error-main'}>
-                            Invalid key, matches strings starting with a letter/underscore, followed by alphanumeric
-                            characters, underscores, hyphens, or dots.
+                            Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                            underscores, hyphens, or dots.
                           </FormHelperText>
                         )}
                         {!item.key.trim() && (
-                          <FormHelperText className={'twc-text-error-main'}>Key is required field</FormHelperText>
+                          <FormHelperText className={'twc-text-error-main'}>Key is required</FormHelperText>
                         )}
                       </FormControl>
                     </Grid>
diff --git a/web/web/src/app/metalakes/metalake/rightContent/CreateTopicDialog.js b/web/web/src/app/metalakes/metalake/rightContent/CreateTopicDialog.js
new file mode 100644
index 00000000000..4f671cc872f
--- /dev/null
+++ b/web/web/src/app/metalakes/metalake/rightContent/CreateTopicDialog.js
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 client'
+
+import { useState, forwardRef, useEffect, Fragment } from 'react'
+
+import {
+  Box,
+  Button,
+  Dialog,
+  DialogActions,
+  DialogContent,
+  Fade,
+  FormControl,
+  FormHelperText,
+  Grid,
+  IconButton,
+  InputLabel,
+  TextField,
+  Typography
+} from '@mui/material'
+
+import Icon from '@/components/Icon'
+
+import { useAppDispatch } from '@/lib/hooks/useStore'
+import { createTopic, updateTopic } from '@/lib/store/metalakes'
+
+import * as yup from 'yup'
+import { useForm, Controller } from 'react-hook-form'
+import { yupResolver } from '@hookform/resolvers/yup'
+
+import { groupBy } from 'lodash-es'
+import { genUpdates } from '@/lib/utils'
+import { nameRegex, nameRegexDesc, keyRegex } from '@/lib/utils/regex'
+import { useSearchParams } from 'next/navigation'
+import { useAppSelector } from '@/lib/hooks/useStore'
+
+const defaultValues = {
+  name: '',
+  comment: '',
+  propItems: []
+}
+
+const schema = yup.object().shape({
+  name: yup.string().required().matches(nameRegex, nameRegexDesc),
+  propItems: yup.array().of(
+    yup.object().shape({
+      required: yup.boolean(),
+      key: yup.string().required(),
+      value: yup.string().when('required', {
+        is: true,
+        then: schema => schema.required()
+      })
+    })
+  )
+})
+
+const Transition = forwardRef(function Transition(props, ref) {
+  return <Fade ref={ref} {...props} />
+})
+
+const CreateTopicDialog = props => {
+  const { open, setOpen, type = 'create', data = {} } = props
+  const searchParams = useSearchParams()
+  const metalake = searchParams.get('metalake')
+  const catalog = searchParams.get('catalog')
+  const schemaName = searchParams.get('schema')
+  const catalogType = searchParams.get('type')
+  const [innerProps, setInnerProps] = useState([])
+  const dispatch = useAppDispatch()
+  const store = useAppSelector(state => state.metalakes)
+  const [cacheData, setCacheData] = useState()
+
+  const {
+    control,
+    reset,
+    watch,
+    setValue,
+    getValues,
+    handleSubmit,
+    trigger,
+    formState: { errors }
+  } = useForm({
+    defaultValues,
+    mode: 'all',
+    resolver: yupResolver(schema)
+  })
+
+  const handleFormChange = ({ index, event }) => {
+    let data = [...innerProps]
+    data[index][event.target.name] = event.target.value
+
+    if (event.target.name === 'key') {
+      const invalidKey = !keyRegex.test(event.target.value)
+      data[index].invalid = invalidKey
+    }
+
+    const nonEmptyKeys = data.filter(item => item.key.trim() !== '')
+    const grouped = groupBy(nonEmptyKeys, 'key')
+    const duplicateKeys = Object.keys(grouped).some(key => grouped[key].length > 1)
+
+    if (duplicateKeys) {
+      data[index].hasDuplicateKey = duplicateKeys
+    } else {
+      data.forEach(it => (it.hasDuplicateKey = false))
+    }
+
+    setInnerProps(data)
+    setValue('propItems', data)
+  }
+
+  const addFields = () => {
+    const duplicateKeys = innerProps
+      .filter(item => item.key.trim() !== '')
+      .some(
+        (item, index, filteredItems) =>
+          filteredItems.findIndex(otherItem => otherItem !== item && otherItem.key.trim() === item.key.trim()) !== -1
+      )
+
+    if (duplicateKeys) {
+      return
+    }
+
+    let newField = { key: '', value: '', required: false }
+
+    setInnerProps([...innerProps, newField])
+    setValue('propItems', [...innerProps, newField])
+  }
+
+  const removeFields = index => {
+    let data = [...innerProps]
+    data.splice(index, 1)
+    setInnerProps(data)
+    setValue('propItems', data)
+  }
+
+  const handleClose = () => {
+    reset()
+    setInnerProps([])
+    setValue('propItems', [])
+    setOpen(false)
+  }
+
+  const handleClickSubmit = e => {
+    e.preventDefault()
+
+    return handleSubmit(onSubmit(getValues()), onError)
+  }
+
+  const onSubmit = data => {
+    const duplicateKeys = innerProps
+      .filter(item => item.key.trim() !== '')
+      .some(
+        (item, index, filteredItems) =>
+          filteredItems.findIndex(otherItem => otherItem !== item && otherItem.key.trim() === item.key.trim()) !== -1
+      )
+
+    const invalidKeys = innerProps.some(i => i.invalid)
+
+    if (duplicateKeys || invalidKeys) {
+      return
+    }
+
+    trigger()
+
+    schema
+      .validate(data)
+      .then(() => {
+        const properties = innerProps.reduce((acc, item) => {
+          acc[item.key] = item.value
+
+          return acc
+        }, {})
+
+        const schemaData = {
+          name: data.name,
+          comment: data.comment,
+          properties
+        }
+
+        if (type === 'create') {
+          dispatch(createTopic({ data: schemaData, metalake, catalog, schema: schemaName, type: catalogType })).then(
+            res => {
+              if (!res.payload?.err) {
+                handleClose()
+              }
+            }
+          )
+        } else {
+          const reqData = { updates: genUpdates(cacheData, schemaData) }
+
+          if (reqData.updates.length !== 0) {
+            dispatch(
+              updateTopic({
+                metalake,
+                catalog,
+                type: catalogType,
+                schema: schemaName,
+                topic: cacheData.name,
+                data: reqData
+              })
+            ).then(res => {
+              if (!res.payload?.err) {
+                handleClose()
+              }
+            })
+          }
+        }
+      })
+      .catch(err => {
+        console.error('valid error', err)
+      })
+  }
+
+  const onError = errors => {
+    console.error('fields error', errors)
+  }
+
+  useEffect(() => {
+    if (open && JSON.stringify(data) !== '{}') {
+      const { properties = {} } = data
+
+      setCacheData(data)
+      setValue('name', data.name)
+      setValue('comment', data.comment)
+
+      const propsItems = Object.entries(properties).map(([key, value]) => {
+        return {
+          key,
+          value
+        }
+      })
+
+      setInnerProps(propsItems)
+      setValue('propItems', propsItems)
+    }
+  }, [open, data, setValue, type])
+
+  return (
+    <Dialog fullWidth maxWidth='sm' scroll='body' TransitionComponent={Transition} open={open} onClose={handleClose}>
+      <form onSubmit={e => handleClickSubmit(e)}>
+        <DialogContent
+          sx={{
+            position: 'relative',
+            pb: theme => `${theme.spacing(8)} !important`,
+            px: theme => [`${theme.spacing(5)} !important`, `${theme.spacing(15)} !important`],
+            pt: theme => [`${theme.spacing(8)} !important`, `${theme.spacing(12.5)} !important`]
+          }}
+        >
+          <IconButton
+            size='small'
+            onClick={() => handleClose()}
+            sx={{ position: 'absolute', right: '1rem', top: '1rem' }}
+          >
+            <Icon icon='bx:x' />
+          </IconButton>
+          <Box sx={{ mb: 8, textAlign: 'center' }}>
+            <Typography variant='h5' sx={{ mb: 3 }}>
+              {type === 'create' ? 'Create' : 'Edit'} Topic
+            </Typography>
+          </Box>
+
+          <Grid container spacing={6}>
+            <Grid item xs={12}>
+              <FormControl fullWidth>
+                <Controller
+                  name='name'
+                  control={control}
+                  rules={{ required: true }}
+                  render={({ field: { value, onChange } }) => (
+                    <TextField
+                      value={value}
+                      label='Name'
+                      onChange={onChange}
+                      placeholder=''
+                      disabled={type === 'update'}
+                      error={Boolean(errors.name)}
+                      data-refer='topic-name-field'
+                    />
+                  )}
+                />
+                {errors.name && <FormHelperText sx={{ color: 'error.main' }}>{errors.name.message}</FormHelperText>}
+              </FormControl>
+            </Grid>
+
+            <Grid item xs={12}>
+              <FormControl fullWidth>
+                <Controller
+                  name='comment'
+                  control={control}
+                  rules={{ required: false }}
+                  render={({ field: { value, onChange } }) => (
+                    <TextField
+                      value={value}
+                      label='Comment'
+                      multiline
+                      rows={2}
+                      onChange={onChange}
+                      placeholder=''
+                      error={Boolean(errors.comment)}
+                      data-refer='topic-comment-field'
+                    />
+                  )}
+                />
+              </FormControl>
+            </Grid>
+
+            <Grid item xs={12} data-refer='topic-props-layout'>
+              <Typography sx={{ mb: 2 }} variant='body2'>
+                Properties
+              </Typography>
+              {innerProps.map((item, index) => {
+                return (
+                  <Fragment key={index}>
+                    <Grid item xs={12} sx={{ '& + &': { mt: 2 } }}>
+                      <FormControl fullWidth>
+                        <Box>
+                          <Box
+                            sx={{ display: 'flex', alignItems: 'center', justifyContent: 'space-between' }}
+                            data-refer={`topic-props-${index}`}
+                          >
+                            <Box>
+                              <TextField
+                                size='small'
+                                name='key'
+                                label='Key'
+                                value={item.key}
+                                disabled={item.disabled || (item.key === 'location' && type === 'update')}
+                                onChange={event => handleFormChange({ index, event })}
+                                error={item.hasDuplicateKey || item.invalid || !item.key.trim()}
+                                data-refer={`props-key-${index}`}
+                              />
+                            </Box>
+                            <Box>
+                              <TextField
+                                size='small'
+                                name='value'
+                                label='Value'
+                                error={item.required && item.value === ''}
+                                value={item.value}
+                                disabled={item.disabled || (item.key === 'location' && type === 'update')}
+                                onChange={event => handleFormChange({ index, event })}
+                                data-refer={`props-value-${index}`}
+                                data-prev-refer={`props-${item.key}`}
+                              />
+                            </Box>
+
+                            {!(item.disabled || (item.key === 'location' && type === 'update')) ? (
+                              <Box sx={{ minWidth: 40 }}>
+                                <IconButton onClick={() => removeFields(index)}>
+                                  <Icon icon='mdi:minus-circle-outline' />
+                                </IconButton>
+                              </Box>
+                            ) : (
+                              <Box sx={{ minWidth: 40 }}></Box>
+                            )}
+                          </Box>
+                        </Box>
+                        <FormHelperText
+                          sx={{
+                            color: item.required && item.value === '' ? 'error.main' : 'text.main',
+                            maxWidth: 'calc(100% - 40px)'
+                          }}
+                        >
+                          {item.description}
+                        </FormHelperText>
+                        {item.hasDuplicateKey && (
+                          <FormHelperText className={'twc-text-error-main'}>Key already exists</FormHelperText>
+                        )}
+                        {item.key && item.invalid && (
+                          <FormHelperText className={'twc-text-error-main'}>
+                            Valid key must starts with a letter/underscore, followed by alphanumeric characters,
+                            underscores, hyphens, or dots.
+                          </FormHelperText>
+                        )}
+                        {!item.key.trim() && (
+                          <FormHelperText className={'twc-text-error-main'}>Key is required</FormHelperText>
+                        )}
+                      </FormControl>
+                    </Grid>
+                  </Fragment>
+                )
+              })}
+            </Grid>
+
+            <Grid item xs={12}>
+              <Button
+                size='small'
+                onClick={addFields}
+                variant='outlined'
+                startIcon={<Icon icon='mdi:plus-circle-outline' />}
+                data-refer='add-topic-props'
+              >
+                Add Property
+              </Button>
+            </Grid>
+          </Grid>
+        </DialogContent>
+        <DialogActions
+          sx={{
+            justifyContent: 'center',
+            px: theme => [`${theme.spacing(5)} !important`, `${theme.spacing(15)} !important`],
+            pb: theme => [`${theme.spacing(5)} !important`, `${theme.spacing(12.5)} !important`]
+          }}
+        >
+          <Button variant='contained' sx={{ mr: 1 }} type='submit' data-refer='handle-submit-topic'>
+            {type === 'create' ? 'Create' : 'Update'}
+          </Button>
+          <Button variant='outlined' onClick={handleClose}>
+            Cancel
+          </Button>
+        </DialogActions>
+      </form>
+    </Dialog>
+  )
+}
+
+export default CreateTopicDialog
diff --git a/web/web/src/app/metalakes/metalake/rightContent/RightContent.js b/web/web/src/app/metalakes/metalake/rightContent/RightContent.js
index a4a59099fe4..8e061f97ac6 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/RightContent.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/RightContent.js
@@ -27,6 +27,7 @@ import MetalakePath from './MetalakePath'
 import CreateCatalogDialog from './CreateCatalogDialog'
 import CreateSchemaDialog from './CreateSchemaDialog'
 import CreateFilesetDialog from './CreateFilesetDialog'
+import CreateTopicDialog from './CreateTopicDialog'
 import CreateTableDialog from './CreateTableDialog'
 import TabsContent from './tabsContent/TabsContent'
 import { useSearchParams } from 'next/navigation'
@@ -36,11 +37,13 @@ const RightContent = () => {
   const [open, setOpen] = useState(false)
   const [openSchema, setOpenSchema] = useState(false)
   const [openFileset, setOpenFileset] = useState(false)
+  const [openTopic, setOpenTopic] = useState(false)
   const [openTable, setOpenTable] = useState(false)
   const searchParams = useSearchParams()
   const [isShowBtn, setBtnVisible] = useState(true)
   const [isShowSchemaBtn, setSchemaBtnVisible] = useState(false)
   const [isShowFilesetBtn, setFilesetBtnVisible] = useState(false)
+  const [isShowTopicBtn, setTopicBtnVisible] = useState(false)
   const [isShowTableBtn, setTableBtnVisible] = useState(false)
   const store = useAppSelector(state => state.metalakes)
 
@@ -56,6 +59,10 @@ const RightContent = () => {
     setOpenFileset(true)
   }
 
+  const handleCreateTopic = () => {
+    setOpenTopic(true)
+  }
+
   const handleCreateTable = () => {
     setOpenTable(true)
   }
@@ -69,10 +76,18 @@ const RightContent = () => {
       paramsSize == 4 &&
       searchParams.has('metalake') &&
       searchParams.has('catalog') &&
-      searchParams.get('type') === 'fileset'
-    searchParams.has('schema')
+      searchParams.get('type') === 'fileset' &&
+      searchParams.has('schema')
     setFilesetBtnVisible(isFilesetList)
 
+    const isTopicList =
+      paramsSize == 4 &&
+      searchParams.has('metalake') &&
+      searchParams.has('catalog') &&
+      searchParams.get('type') === 'messaging' &&
+      searchParams.has('schema')
+    setTopicBtnVisible(isTopicList)
+
     if (store.catalogs.length) {
       const currentCatalog = store.catalogs.filter(ca => ca.name === searchParams.get('catalog'))[0]
 
@@ -83,15 +98,16 @@ const RightContent = () => {
         searchParams.has('type') &&
         !['lakehouse-hudi', 'kafka'].includes(currentCatalog?.provider)
       setSchemaBtnVisible(isSchemaList)
-    }
 
-    const isTableList =
-      paramsSize == 4 &&
-      searchParams.has('metalake') &&
-      searchParams.has('catalog') &&
-      searchParams.get('type') === 'relational' &&
-      searchParams.has('schema')
-    setTableBtnVisible(isTableList)
+      const isTableList =
+        paramsSize == 4 &&
+        searchParams.has('metalake') &&
+        searchParams.has('catalog') &&
+        searchParams.get('type') === 'relational' &&
+        searchParams.has('schema') &&
+        'lakehouse-hudi' !== currentCatalog?.provider
+      setTableBtnVisible(isTableList)
+    }
   }, [searchParams, store.catalogs, store.catalogs.length])
 
   return (
@@ -155,6 +171,20 @@ const RightContent = () => {
             <CreateFilesetDialog open={openFileset} setOpen={setOpenFileset} />
           </Box>
         )}
+        {isShowTopicBtn && (
+          <Box className={`twc-flex twc-items-center`}>
+            <Button
+              variant='contained'
+              startIcon={<Icon icon='mdi:plus-box' />}
+              onClick={handleCreateTopic}
+              sx={{ width: 200 }}
+              data-refer='create-topic-btn'
+            >
+              Create Topic
+            </Button>
+            <CreateTopicDialog open={openTopic} setOpen={setOpenTopic} />
+          </Box>
+        )}
         {isShowTableBtn && (
           <Box className={`twc-flex twc-items-center`}>
             <Button
diff --git a/web/web/src/app/metalakes/metalake/rightContent/tabsContent/tableView/TableView.js b/web/web/src/app/metalakes/metalake/rightContent/tabsContent/tableView/TableView.js
index f414ccfeccb..a2a73c1ecfb 100644
--- a/web/web/src/app/metalakes/metalake/rightContent/tabsContent/tableView/TableView.js
+++ b/web/web/src/app/metalakes/metalake/rightContent/tabsContent/tableView/TableView.js
@@ -42,16 +42,25 @@ import ConfirmDeleteDialog from '@/components/ConfirmDeleteDialog'
 import CreateCatalogDialog from '../../CreateCatalogDialog'
 import CreateSchemaDialog from '../../CreateSchemaDialog'
 import CreateFilesetDialog from '../../CreateFilesetDialog'
+import CreateTopicDialog from '../../CreateTopicDialog'
 import CreateTableDialog from '../../CreateTableDialog'
 
 import { useAppSelector, useAppDispatch } from '@/lib/hooks/useStore'
-import { deleteCatalog, deleteFileset, deleteSchema, deleteTable, setCatalogInUse } from '@/lib/store/metalakes'
+import {
+  deleteCatalog,
+  deleteFileset,
+  deleteTopic,
+  deleteSchema,
+  deleteTable,
+  setCatalogInUse
+} from '@/lib/store/metalakes'
 
 import { to } from '@/lib/utils'
 import { getCatalogDetailsApi, switchInUseApi } from '@/lib/api/catalogs'
 import { getSchemaDetailsApi } from '@/lib/api/schemas'
 import { useSearchParams } from 'next/navigation'
 import { getFilesetDetailsApi } from '@/lib/api/filesets'
+import { getTopicDetailsApi } from '@/lib/api/topics'
 import { getTableDetailsApi } from '@/lib/api/tables'
 
 const fonts = Inconsolata({ subsets: ['latin'] })
@@ -105,16 +114,20 @@ const TableView = () => {
   const [openDialog, setOpenDialog] = useState(false)
   const [openSchemaDialog, setOpenSchemaDialog] = useState(false)
   const [openFilesetDialog, setOpenFilesetDialog] = useState(false)
+  const [openTopicDialog, setOpenTopicDialog] = useState(false)
   const [openTableDialog, setOpenTableDialog] = useState(false)
   const [dialogData, setDialogData] = useState({})
   const [dialogType, setDialogType] = useState('create')
-  const [isHideSchemaEdit, setIsHideSchemaEdit] = useState(true)
+  const [isHideEdit, setIsHideEdit] = useState(true)
 
   useEffect(() => {
     if (store.catalogs.length) {
       const currentCatalog = store.catalogs.filter(ca => ca.name === catalog)[0]
-      const isHideSchemaAction = ['lakehouse-hudi', 'kafka'].includes(currentCatalog?.provider) && paramsSize == 3
-      setIsHideSchemaEdit(isHideSchemaAction)
+
+      const isHideAction =
+        (['lakehouse-hudi', 'kafka'].includes(currentCatalog?.provider) && paramsSize == 3) ||
+        (currentCatalog?.provider === 'lakehouse-hudi' && paramsSize == 4)
+      setIsHideEdit(isHideAction)
     }
   }, [store.catalogs, store.catalogs.length, paramsSize, catalog])
 
@@ -300,7 +313,7 @@ const TableView = () => {
             <ViewIcon viewBox='0 0 24 22' />
           </IconButton>
 
-          {!isHideSchemaEdit && (
+          {!isHideEdit && (
             <IconButton
               title='Edit'
               size='small'
@@ -312,7 +325,7 @@ const TableView = () => {
             </IconButton>
           )}
 
-          {!isHideSchemaEdit && (
+          {!isHideEdit && (
             <IconButton
               title='Delete'
               size='small'
@@ -502,6 +515,16 @@ const TableView = () => {
         setOpenDrawer(true)
         break
       }
+      case 'topic': {
+        const [err, res] = await to(getTopicDetailsApi({ metalake, catalog, schema, topic: row.name }))
+        if (err || !res) {
+          throw new Error(err)
+        }
+
+        setDrawerData(res.topic)
+        setOpenDrawer(true)
+        break
+      }
       case 'table': {
         const [err, res] = await to(getTableDetailsApi({ metalake, catalog, schema, table: row.name }))
         if (err || !res) {
@@ -560,6 +583,19 @@ const TableView = () => {
         }
         break
       }
+      case 'topic': {
+        if (metalake && catalog && schema) {
+          const [err, res] = await to(getTopicDetailsApi({ metalake, catalog, schema, topic: data.row?.name }))
+          if (err || !res) {
+            throw new Error(err)
+          }
+
+          setDialogType('update')
+          setDialogData(res.topic)
+          setOpenTopicDialog(true)
+        }
+        break
+      }
       case 'table': {
         if (metalake && catalog && schema) {
           const [err, res] = await to(getTableDetailsApi({ metalake, catalog, schema, table: data.row?.name }))
@@ -600,6 +636,9 @@ const TableView = () => {
         case 'fileset':
           dispatch(deleteFileset({ metalake, catalog, type, schema, fileset: confirmCacheData.name }))
           break
+        case 'topic':
+          dispatch(deleteTopic({ metalake, catalog, type, schema, topic: confirmCacheData.name }))
+          break
         case 'table':
           dispatch(deleteTable({ metalake, catalog, type, schema, table: confirmCacheData.name }))
           break
@@ -628,6 +667,11 @@ const TableView = () => {
         searchParams.has('catalog') &&
         searchParams.get('type') === 'fileset' &&
         searchParams.has('schema')) ||
+      (paramsSize == 4 &&
+        searchParams.has('metalake') &&
+        searchParams.has('catalog') &&
+        searchParams.get('type') === 'messaging' &&
+        searchParams.has('schema')) ||
       (paramsSize == 4 &&
         searchParams.has('metalake') &&
         searchParams.has('catalog') &&
@@ -687,6 +731,8 @@ const TableView = () => {
         type={dialogType}
       />
 
+      <CreateTopicDialog open={openTopicDialog} setOpen={setOpenTopicDialog} data={dialogData} type={dialogType} />
+
       <CreateTableDialog open={openTableDialog} setOpen={setOpenTableDialog} data={dialogData} type={dialogType} />
     </Box>
   )
diff --git a/web/web/src/lib/api/topics/index.js b/web/web/src/lib/api/topics/index.js
index 2ba86ae18a4..24f8afd81f1 100644
--- a/web/web/src/lib/api/topics/index.js
+++ b/web/web/src/lib/api/topics/index.js
@@ -27,7 +27,13 @@ const Apis = {
   GET_DETAIL: ({ metalake, catalog, schema, topic }) =>
     `/api/metalakes/${encodeURIComponent(metalake)}/catalogs/${encodeURIComponent(
       catalog
-    )}/schemas/${encodeURIComponent(schema)}/topics/${encodeURIComponent(topic)}`
+    )}/schemas/${encodeURIComponent(schema)}/topics/${encodeURIComponent(topic)}`,
+  CREATE: ({ metalake, catalog, schema }) =>
+    `/api/metalakes/${encodeURIComponent(metalake)}/catalogs/${encodeURIComponent(catalog)}/schemas/${encodeURIComponent(schema)}/topics`,
+  UPDATE: ({ metalake, catalog, schema, topic }) =>
+    `/api/metalakes/${encodeURIComponent(metalake)}/catalogs/${encodeURIComponent(catalog)}/schemas/${encodeURIComponent(schema)}/topics/${encodeURIComponent(topic)}`,
+  DELETE: ({ metalake, catalog, schema, topic }) =>
+    `/api/metalakes/${encodeURIComponent(metalake)}/catalogs/${encodeURIComponent(catalog)}/schemas/${encodeURIComponent(schema)}/topics/${encodeURIComponent(topic)}`
 }
 
 export const getTopicsApi = params => {
@@ -41,3 +47,15 @@ export const getTopicDetailsApi = ({ metalake, catalog, schema, topic }) => {
     url: `${Apis.GET_DETAIL({ metalake, catalog, schema, topic })}`
   })
 }
+
+export const createTopicApi = ({ metalake, catalog, schema, data }) => {
+  return defHttp.post({ url: `${Apis.CREATE({ metalake, catalog, schema })}`, data })
+}
+
+export const updateTopicApi = ({ metalake, catalog, schema, topic, data }) => {
+  return defHttp.put({ url: `${Apis.UPDATE({ metalake, catalog, schema, topic })}`, data })
+}
+
+export const deleteTopicApi = ({ metalake, catalog, schema, topic }) => {
+  return defHttp.delete({ url: `${Apis.DELETE({ metalake, catalog, schema, topic })}` })
+}
diff --git a/web/web/src/lib/store/metalakes/index.js b/web/web/src/lib/store/metalakes/index.js
index 6b8fedf3296..3d4ad454d87 100644
--- a/web/web/src/lib/store/metalakes/index.js
+++ b/web/web/src/lib/store/metalakes/index.js
@@ -54,7 +54,7 @@ import {
   updateFilesetApi,
   deleteFilesetApi
 } from '@/lib/api/filesets'
-import { getTopicsApi, getTopicDetailsApi } from '@/lib/api/topics'
+import { getTopicsApi, getTopicDetailsApi, createTopicApi, updateTopicApi, deleteTopicApi } from '@/lib/api/topics'
 
 export const fetchMetalakes = createAsyncThunk('appMetalakes/fetchMetalakes', async (params, { getState }) => {
   const [err, res] = await to(getMetalakesApi())
@@ -1104,6 +1104,67 @@ export const getTopicDetails = createAsyncThunk(
   }
 )
 
+export const createTopic = createAsyncThunk(
+  'appMetalakes/createTopic',
+  async ({ data, metalake, catalog, type, schema }, { dispatch }) => {
+    dispatch(setTableLoading(true))
+    const [err, res] = await to(createTopicApi({ data, metalake, catalog, schema }))
+    dispatch(setTableLoading(false))
+
+    if (err || !res) {
+      return { err: true }
+    }
+
+    const { topic: topicItem } = res
+
+    const topicData = {
+      ...topicItem,
+      node: 'topic',
+      id: `{{${metalake}}}{{${catalog}}}{{${type}}}{{${schema}}}{{${topicItem.name}}}`,
+      key: `{{${metalake}}}{{${catalog}}}{{${type}}}{{${schema}}}{{${topicItem.name}}}`,
+      path: `?${new URLSearchParams({ metalake, catalog, type, schema, topic: topicItem.name }).toString()}`,
+      name: topicItem.name,
+      title: topicItem.name,
+      tables: [],
+      children: []
+    }
+
+    dispatch(fetchTopics({ metalake, catalog, schema, type, init: true }))
+
+    return topicData
+  }
+)
+
+export const updateTopic = createAsyncThunk(
+  'appMetalakes/updateTopic',
+  async ({ metalake, catalog, type, schema, topic, data }, { dispatch }) => {
+    const [err, res] = await to(updateTopicApi({ metalake, catalog, schema, topic, data }))
+    if (err || !res) {
+      return { err: true }
+    }
+    dispatch(fetchTopics({ metalake, catalog, type, schema, init: true }))
+
+    return res.catalog
+  }
+)
+
+export const deleteTopic = createAsyncThunk(
+  'appMetalakes/deleteTopic',
+  async ({ metalake, catalog, type, schema, topic }, { dispatch }) => {
+    dispatch(setTableLoading(true))
+    const [err, res] = await to(deleteTopicApi({ metalake, catalog, schema, topic }))
+    dispatch(setTableLoading(false))
+
+    if (err || !res) {
+      throw new Error(err)
+    }
+
+    dispatch(fetchTopics({ metalake, catalog, type, schema, page: 'topics', init: true }))
+
+    return res
+  }
+)
+
 export const appMetalakesSlice = createSlice({
   name: 'appMetalakes',
   initialState: {
@@ -1346,6 +1407,21 @@ export const appMetalakesSlice = createSlice({
         toast.error(action.error.message)
       }
     })
+    builder.addCase(createTable.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(updateTable.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(deleteTable.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
     builder.addCase(fetchFilesets.fulfilled, (state, action) => {
       state.filesets = action.payload.filesets
       if (action.payload.init) {
@@ -1366,6 +1442,21 @@ export const appMetalakesSlice = createSlice({
         toast.error(action.error.message)
       }
     })
+    builder.addCase(createFileset.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(updateFileset.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(deleteFileset.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
     builder.addCase(fetchTopics.fulfilled, (state, action) => {
       state.topics = action.payload.topics
       if (action.payload.init) {
@@ -1386,6 +1477,21 @@ export const appMetalakesSlice = createSlice({
         toast.error(action.error.message)
       }
     })
+    builder.addCase(createTopic.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(updateTopic.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
+    builder.addCase(deleteTopic.rejected, (state, action) => {
+      if (!action.error.message.includes('CanceledError')) {
+        toast.error(action.error.message)
+      }
+    })
   }
 })