diff --git a/ui/src/components/DataSourceFormModal.tsx b/ui/src/components/DataSourceFormModal.tsx
index d9fa822d381..328c614d3d2 100644
--- a/ui/src/components/DataSourceFormModal.tsx
+++ b/ui/src/components/DataSourceFormModal.tsx
@@ -7,11 +7,16 @@ import {
   EuiHorizontalRule,
   EuiText,
   EuiCallOut,
+  EuiFlexGroup,
+  EuiFlexItem,
+  EuiPanel,
+  EuiTextArea,
+  EuiTitle,
 } from "@elastic/eui";
 import { feast } from "../protos";
 import FormModal from "./forms/FormModal";
 import TagsEditor, { TagEntry } from "./forms/TagsEditor";
-import NameDescriptionOwnerFields from "./forms/NameDescriptionOwnerFields";
+import { DATA_SOURCE_TYPES } from "../pages/data-sources/DataSourceCatalog";
 
 const SOURCE_TYPE_OPTIONS = [
   {
@@ -30,13 +35,25 @@ const SOURCE_TYPE_OPTIONS = [
     value: String(feast.core.DataSource.SourceType.BATCH_REDSHIFT),
     text: "Redshift",
   },
+  {
+    value: String(feast.core.DataSource.SourceType.BATCH_SPARK),
+    text: "Spark",
+  },
+  {
+    value: String(feast.core.DataSource.SourceType.BATCH_TRINO),
+    text: "Trino",
+  },
+  {
+    value: String(feast.core.DataSource.SourceType.BATCH_ATHENA),
+    text: "AWS Athena",
+  },
   {
     value: String(feast.core.DataSource.SourceType.STREAM_KAFKA),
     text: "Kafka",
   },
   {
-    value: String(feast.core.DataSource.SourceType.BATCH_SPARK),
-    text: "Spark",
+    value: String(feast.core.DataSource.SourceType.STREAM_KINESIS),
+    text: "AWS Kinesis",
   },
   {
     value: String(feast.core.DataSource.SourceType.REQUEST_SOURCE),
@@ -46,6 +63,17 @@ const SOURCE_TYPE_OPTIONS = [
     value: String(feast.core.DataSource.SourceType.PUSH_SOURCE),
     text: "Push Source",
   },
+  {
+    value: String(feast.core.DataSource.SourceType.CUSTOM_SOURCE),
+    text: "Custom Source",
+  },
+  { value: "RAY_SOURCE", text: "Ray" },
+  { value: "POSTGRES_SOURCE", text: "PostgreSQL" },
+  { value: "MONGODB_SOURCE", text: "MongoDB" },
+  { value: "CLICKHOUSE_SOURCE", text: "ClickHouse" },
+  { value: "MSSQL_SOURCE", text: "SQL Server" },
+  { value: "ORACLE_SOURCE", text: "Oracle" },
+  { value: "COUCHBASE_SOURCE", text: "Couchbase" },
 ];
 
 interface DataSourceFormData {
@@ -69,6 +97,33 @@ interface DataSourceFormData {
   kafkaTopic: string;
   sparkTable: string;
   sparkPath: string;
+  kinesisRegion: string;
+  kinesisStreamName: string;
+  trinoTable: string;
+  trinoQuery: string;
+  athenaTable: string;
+  athenaQuery: string;
+  athenaDatabase: string;
+  athenaDataSource: string;
+  customSourceClassName: string;
+  customSourceConfig: string;
+  // Contrib source fields
+  rayReaderType: string;
+  rayPath: string;
+  rayReaderOptions: string;
+  postgresTable: string;
+  postgresQuery: string;
+  mongodbCollection: string;
+  clickhouseTable: string;
+  clickhouseQuery: string;
+  mssqlTable: string;
+  mssqlConnectionStr: string;
+  oracleTable: string;
+  oracleConnectionStr: string;
+  couchbaseDatabase: string;
+  couchbaseScope: string;
+  couchbaseCollection: string;
+  couchbaseQuery: string;
 }
 
 interface DataSourceFormModalProps {
@@ -101,6 +156,32 @@ const EMPTY_FORM: DataSourceFormData = {
   kafkaTopic: "",
   sparkTable: "",
   sparkPath: "",
+  kinesisRegion: "",
+  kinesisStreamName: "",
+  trinoTable: "",
+  trinoQuery: "",
+  athenaTable: "",
+  athenaQuery: "",
+  athenaDatabase: "",
+  athenaDataSource: "",
+  customSourceClassName: "",
+  customSourceConfig: "",
+  rayReaderType: "parquet",
+  rayPath: "",
+  rayReaderOptions: "",
+  postgresTable: "",
+  postgresQuery: "",
+  mongodbCollection: "",
+  clickhouseTable: "",
+  clickhouseQuery: "",
+  mssqlTable: "",
+  mssqlConnectionStr: "",
+  oracleTable: "",
+  oracleConnectionStr: "",
+  couchbaseDatabase: "",
+  couchbaseScope: "",
+  couchbaseCollection: "",
+  couchbaseQuery: "",
 };
 
 const BATCH_SOURCE_TYPES = new Set([
@@ -109,8 +190,31 @@ const BATCH_SOURCE_TYPES = new Set([
   String(feast.core.DataSource.SourceType.BATCH_SNOWFLAKE),
   String(feast.core.DataSource.SourceType.BATCH_REDSHIFT),
   String(feast.core.DataSource.SourceType.BATCH_SPARK),
+  String(feast.core.DataSource.SourceType.BATCH_TRINO),
+  String(feast.core.DataSource.SourceType.BATCH_ATHENA),
+  "RAY_SOURCE",
+  "POSTGRES_SOURCE",
+  "MONGODB_SOURCE",
+  "CLICKHOUSE_SOURCE",
+  "MSSQL_SOURCE",
+  "ORACLE_SOURCE",
+  "COUCHBASE_SOURCE",
 ]);
 
+const RAY_READER_OPTIONS = [
+  { value: "parquet", text: "Parquet" },
+  { value: "csv", text: "CSV" },
+  { value: "json", text: "JSON" },
+  { value: "text", text: "Text" },
+  { value: "images", text: "Images" },
+  { value: "binary_files", text: "Binary Files" },
+  { value: "tfrecords", text: "TFRecords" },
+  { value: "webdataset", text: "WebDataset" },
+  { value: "huggingface", text: "HuggingFace" },
+  { value: "mongo", text: "MongoDB (via Ray)" },
+  { value: "sql", text: "SQL (via Ray)" },
+];
+
 const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
   onClose,
   onSubmit,
@@ -125,13 +229,12 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
   const [errors, setErrors] = useState<Record<string, string>>({});
   const [submitted, setSubmitted] = useState(false);
 
-  useEffect(() => {
-    if (initialData) {
-      setFormData(initialData);
-    }
-  }, [initialData]);
-
   const isBatchSource = BATCH_SOURCE_TYPES.has(formData.sourceType);
+  const isPreselected = !!initialData?.sourceType;
+
+  const catalogEntry = DATA_SOURCE_TYPES.find(
+    (ds) => ds.sourceType === formData.sourceType,
+  );
 
   const validate = (): boolean => {
     const newErrors: Record<string, string> = {};
@@ -144,7 +247,6 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
         "Must start with a letter or underscore, and contain only letters, numbers, and underscores.";
     }
 
-    // Source-type-specific required fields
     if (st === String(feast.core.DataSource.SourceType.BATCH_FILE)) {
       if (!formData.fileUri.trim()) {
         newErrors.fileUri = "File URI is required.";
@@ -180,6 +282,19 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
         newErrors.sparkTable =
           "Either a table reference or a path is required for Spark.";
       }
+    } else if (st === String(feast.core.DataSource.SourceType.BATCH_TRINO)) {
+      if (!formData.trinoTable.trim() && !formData.trinoQuery.trim()) {
+        newErrors.trinoTable =
+          "Either a table reference or a query is required for Trino.";
+      }
+    } else if (st === String(feast.core.DataSource.SourceType.BATCH_ATHENA)) {
+      if (!formData.athenaTable.trim() && !formData.athenaQuery.trim()) {
+        newErrors.athenaTable =
+          "Either a table reference or a query is required for Athena.";
+      }
+      if (!formData.athenaDatabase.trim()) {
+        newErrors.athenaDatabase = "Database is required for Athena.";
+      }
     } else if (st === String(feast.core.DataSource.SourceType.STREAM_KAFKA)) {
       if (!formData.kafkaBootstrapServers.trim()) {
         newErrors.kafkaBootstrapServers = "Bootstrap servers are required.";
@@ -194,12 +309,56 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
       if (!formData.kafkaTopic.trim()) {
         newErrors.kafkaTopic = "Topic is required.";
       }
+    } else if (st === String(feast.core.DataSource.SourceType.STREAM_KINESIS)) {
+      if (!formData.kinesisRegion.trim()) {
+        newErrors.kinesisRegion = "AWS region is required.";
+      }
+      if (!formData.kinesisStreamName.trim()) {
+        newErrors.kinesisStreamName = "Stream name is required.";
+      }
+    } else if (st === String(feast.core.DataSource.SourceType.CUSTOM_SOURCE)) {
+      if (!formData.customSourceClassName.trim()) {
+        newErrors.customSourceClassName = "Class name is required.";
+      }
+    } else if (st === "RAY_SOURCE") {
+      if (
+        !formData.rayPath.trim() &&
+        !["huggingface", "mongo", "sql"].includes(formData.rayReaderType)
+      ) {
+        newErrors.rayPath = "Path is required for this reader type.";
+      }
+    } else if (st === "POSTGRES_SOURCE") {
+      if (!formData.postgresTable.trim() && !formData.postgresQuery.trim()) {
+        newErrors.postgresTable = "Either a table or query is required.";
+      }
+    } else if (st === "CLICKHOUSE_SOURCE") {
+      if (
+        !formData.clickhouseTable.trim() &&
+        !formData.clickhouseQuery.trim()
+      ) {
+        newErrors.clickhouseTable = "Either a table or query is required.";
+      }
+    } else if (st === "MSSQL_SOURCE") {
+      if (!formData.mssqlTable.trim()) {
+        newErrors.mssqlTable = "Table reference is required.";
+      }
+    } else if (st === "ORACLE_SOURCE") {
+      if (!formData.oracleTable.trim()) {
+        newErrors.oracleTable = "Table reference is required.";
+      }
+    } else if (st === "COUCHBASE_SOURCE") {
+      if (
+        !formData.couchbaseCollection.trim() &&
+        !formData.couchbaseQuery.trim()
+      ) {
+        newErrors.couchbaseCollection =
+          "Either a collection or query is required.";
+      }
     }
 
-    // Timestamp field required for batch sources (needed for point-in-time correctness)
     if (isBatchSource && !formData.timestampField.trim()) {
       newErrors.timestampField =
-        "Timestamp field is required for batch sources. It is used for point-in-time correct feature retrieval.";
+        "Timestamp field is required for batch sources.";
     } else if (
       formData.timestampField.trim() &&
       !/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(formData.timestampField.trim())
@@ -242,25 +401,65 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
     }
   };
 
+  const renderFileSourceFields = () => (
+    <EuiFormRow
+      label="File Path / URI"
+      isInvalid={!!errors.fileUri}
+      error={errors.fileUri}
+      helpText="Path to the data file accessible by the Feast server (e.g. s3://bucket/path/data.parquet, gs://bucket/data.csv, file:///mnt/data/features.parquet)."
+    >
+      <EuiFieldText
+        value={formData.fileUri}
+        onChange={(e) => updateField("fileUri", e.target.value)}
+        isInvalid={!!errors.fileUri}
+        placeholder="s3://bucket/path/to/data.parquet"
+      />
+    </EuiFormRow>
+  );
+
+  const renderSourceTypeHeader = () => {
+    if (!isPreselected || !catalogEntry) return null;
+
+    const IconComponent = catalogEntry.icon;
+    return (
+      <EuiPanel
+        color="subdued"
+        paddingSize="m"
+        hasBorder={false}
+        style={{ borderLeft: `4px solid ${catalogEntry.color}` }}
+      >
+        <EuiFlexGroup alignItems="center" gutterSize="m" responsive={false}>
+          <EuiFlexItem grow={false}>
+            <div
+              style={{
+                width: 36,
+                height: 36,
+                display: "flex",
+                alignItems: "center",
+                justifyContent: "center",
+              }}
+            >
+              <IconComponent width={28} height={28} />
+            </div>
+          </EuiFlexItem>
+          <EuiFlexItem>
+            <EuiText size="s">
+              <strong>{catalogEntry.name}</strong>
+            </EuiText>
+            <EuiText size="xs" color="subdued">
+              {catalogEntry.description}
+            </EuiText>
+          </EuiFlexItem>
+        </EuiFlexGroup>
+      </EuiPanel>
+    );
+  };
+
   const renderSourceTypeFields = () => {
     const st = formData.sourceType;
 
     if (st === String(feast.core.DataSource.SourceType.BATCH_FILE)) {
-      return (
-        <EuiFormRow
-          label="File URI"
-          isInvalid={!!errors.fileUri}
-          error={errors.fileUri}
-          helpText="Path to parquet or CSV file(s). Supports s3://, gs://, and file:// schemes."
-        >
-          <EuiFieldText
-            value={formData.fileUri}
-            onChange={(e) => updateField("fileUri", e.target.value)}
-            isInvalid={!!errors.fileUri}
-            placeholder="s3://bucket/path/to/data.parquet"
-          />
-        </EuiFormRow>
-      );
+      return renderFileSourceFields();
     }
 
     if (st === String(feast.core.DataSource.SourceType.BATCH_BIGQUERY)) {
@@ -280,13 +479,14 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
             />
           </EuiFormRow>
           <EuiFormRow
-            label="Query"
-            helpText="Optional SQL query — use instead of a fixed table reference."
+            label="Query (optional)"
+            helpText="SQL query as an alternative to a fixed table."
           >
-            <EuiFieldText
+            <EuiTextArea
               value={formData.bigqueryQuery}
               onChange={(e) => updateField("bigqueryQuery", e.target.value)}
-              placeholder="SELECT * FROM `project.dataset.table`"
+              placeholder="SELECT * FROM `project.dataset.table` WHERE ..."
+              rows={3}
             />
           </EuiFormRow>
         </>
@@ -296,25 +496,35 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
     if (st === String(feast.core.DataSource.SourceType.BATCH_SNOWFLAKE)) {
       return (
         <>
-          <EuiFormRow
-            label="Database"
-            isInvalid={!!errors.snowflakeDatabase}
-            error={errors.snowflakeDatabase}
-          >
-            <EuiFieldText
-              value={formData.snowflakeDatabase}
-              onChange={(e) => updateField("snowflakeDatabase", e.target.value)}
-              isInvalid={!!errors.snowflakeDatabase}
-              placeholder="MY_DATABASE"
-            />
-          </EuiFormRow>
-          <EuiFormRow label="Schema">
-            <EuiFieldText
-              value={formData.snowflakeSchema}
-              onChange={(e) => updateField("snowflakeSchema", e.target.value)}
-              placeholder="PUBLIC"
-            />
-          </EuiFormRow>
+          <EuiFlexGroup gutterSize="m">
+            <EuiFlexItem>
+              <EuiFormRow
+                label="Database"
+                isInvalid={!!errors.snowflakeDatabase}
+                error={errors.snowflakeDatabase}
+              >
+                <EuiFieldText
+                  value={formData.snowflakeDatabase}
+                  onChange={(e) =>
+                    updateField("snowflakeDatabase", e.target.value)
+                  }
+                  isInvalid={!!errors.snowflakeDatabase}
+                  placeholder="MY_DATABASE"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+            <EuiFlexItem>
+              <EuiFormRow label="Schema">
+                <EuiFieldText
+                  value={formData.snowflakeSchema}
+                  onChange={(e) =>
+                    updateField("snowflakeSchema", e.target.value)
+                  }
+                  placeholder="PUBLIC"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+          </EuiFlexGroup>
           <EuiFormRow
             label="Table"
             isInvalid={!!errors.snowflakeTable}
@@ -334,25 +544,35 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
     if (st === String(feast.core.DataSource.SourceType.BATCH_REDSHIFT)) {
       return (
         <>
-          <EuiFormRow
-            label="Database"
-            isInvalid={!!errors.redshiftDatabase}
-            error={errors.redshiftDatabase}
-          >
-            <EuiFieldText
-              value={formData.redshiftDatabase}
-              onChange={(e) => updateField("redshiftDatabase", e.target.value)}
-              isInvalid={!!errors.redshiftDatabase}
-              placeholder="my_database"
-            />
-          </EuiFormRow>
-          <EuiFormRow label="Schema">
-            <EuiFieldText
-              value={formData.redshiftSchema}
-              onChange={(e) => updateField("redshiftSchema", e.target.value)}
-              placeholder="public"
-            />
-          </EuiFormRow>
+          <EuiFlexGroup gutterSize="m">
+            <EuiFlexItem>
+              <EuiFormRow
+                label="Database"
+                isInvalid={!!errors.redshiftDatabase}
+                error={errors.redshiftDatabase}
+              >
+                <EuiFieldText
+                  value={formData.redshiftDatabase}
+                  onChange={(e) =>
+                    updateField("redshiftDatabase", e.target.value)
+                  }
+                  isInvalid={!!errors.redshiftDatabase}
+                  placeholder="my_database"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+            <EuiFlexItem>
+              <EuiFormRow label="Schema">
+                <EuiFieldText
+                  value={formData.redshiftSchema}
+                  onChange={(e) =>
+                    updateField("redshiftSchema", e.target.value)
+                  }
+                  placeholder="public"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+          </EuiFlexGroup>
           <EuiFormRow
             label="Table"
             isInvalid={!!errors.redshiftTable}
@@ -376,7 +596,7 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
             label="Bootstrap Servers"
             isInvalid={!!errors.kafkaBootstrapServers}
             error={errors.kafkaBootstrapServers}
-            helpText="Comma-separated list of broker host:port pairs."
+            helpText="Comma-separated host:port pairs."
           >
             <EuiFieldText
               value={formData.kafkaBootstrapServers}
@@ -410,7 +630,7 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
             label="Table"
             isInvalid={!!errors.sparkTable}
             error={errors.sparkTable}
-            helpText="Spark catalog table reference (catalog.database.table). Provide either table or path."
+            helpText="Spark catalog table (catalog.database.table). Provide either table or path."
           >
             <EuiFieldText
               value={formData.sparkTable}
@@ -420,8 +640,8 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
             />
           </EuiFormRow>
           <EuiFormRow
-            label="Path"
-            helpText="Alternative to table: path to data files (s3://, gs://, hdfs://)."
+            label="Path (optional)"
+            helpText="Alternative: direct path to data files."
           >
             <EuiFieldText
               value={formData.sparkPath}
@@ -433,37 +653,428 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
       );
     }
 
+    if (st === String(feast.core.DataSource.SourceType.BATCH_TRINO)) {
+      return (
+        <>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.trinoTable}
+            error={errors.trinoTable}
+            helpText="Trino catalog table (catalog.schema.table). Provide either table or query."
+          >
+            <EuiFieldText
+              value={formData.trinoTable}
+              onChange={(e) => updateField("trinoTable", e.target.value)}
+              isInvalid={!!errors.trinoTable}
+              placeholder="catalog.schema.table"
+            />
+          </EuiFormRow>
+          <EuiFormRow label="Query (optional)">
+            <EuiTextArea
+              value={formData.trinoQuery}
+              onChange={(e) => updateField("trinoQuery", e.target.value)}
+              placeholder="SELECT * FROM catalog.schema.table"
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === String(feast.core.DataSource.SourceType.BATCH_ATHENA)) {
+      return (
+        <>
+          <EuiFlexGroup gutterSize="m">
+            <EuiFlexItem>
+              <EuiFormRow
+                label="Database"
+                isInvalid={!!errors.athenaDatabase}
+                error={errors.athenaDatabase}
+              >
+                <EuiFieldText
+                  value={formData.athenaDatabase}
+                  onChange={(e) =>
+                    updateField("athenaDatabase", e.target.value)
+                  }
+                  isInvalid={!!errors.athenaDatabase}
+                  placeholder="my_database"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+            <EuiFlexItem>
+              <EuiFormRow label="Data Source" helpText="Athena catalog name.">
+                <EuiFieldText
+                  value={formData.athenaDataSource}
+                  onChange={(e) =>
+                    updateField("athenaDataSource", e.target.value)
+                  }
+                  placeholder="AwsDataCatalog"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+          </EuiFlexGroup>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.athenaTable}
+            error={errors.athenaTable}
+            helpText="Provide either a table name or a query."
+          >
+            <EuiFieldText
+              value={formData.athenaTable}
+              onChange={(e) => updateField("athenaTable", e.target.value)}
+              isInvalid={!!errors.athenaTable}
+              placeholder="my_table"
+            />
+          </EuiFormRow>
+          <EuiFormRow label="Query (optional)">
+            <EuiTextArea
+              value={formData.athenaQuery}
+              onChange={(e) => updateField("athenaQuery", e.target.value)}
+              placeholder="SELECT * FROM my_table"
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === String(feast.core.DataSource.SourceType.STREAM_KINESIS)) {
+      return (
+        <>
+          <EuiFormRow
+            label="AWS Region"
+            isInvalid={!!errors.kinesisRegion}
+            error={errors.kinesisRegion}
+          >
+            <EuiFieldText
+              value={formData.kinesisRegion}
+              onChange={(e) => updateField("kinesisRegion", e.target.value)}
+              isInvalid={!!errors.kinesisRegion}
+              placeholder="us-east-1"
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Stream Name"
+            isInvalid={!!errors.kinesisStreamName}
+            error={errors.kinesisStreamName}
+          >
+            <EuiFieldText
+              value={formData.kinesisStreamName}
+              onChange={(e) => updateField("kinesisStreamName", e.target.value)}
+              isInvalid={!!errors.kinesisStreamName}
+              placeholder="my-feature-stream"
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === String(feast.core.DataSource.SourceType.CUSTOM_SOURCE)) {
+      return (
+        <>
+          <EuiFormRow
+            label="Data Source Class"
+            isInvalid={!!errors.customSourceClassName}
+            error={errors.customSourceClassName}
+            helpText="Fully qualified Python class name."
+          >
+            <EuiFieldText
+              value={formData.customSourceClassName}
+              onChange={(e) =>
+                updateField("customSourceClassName", e.target.value)
+              }
+              isInvalid={!!errors.customSourceClassName}
+              placeholder="mymodule.MyCustomDataSource"
+            />
+          </EuiFormRow>
+          <EuiFormRow label="Configuration (JSON, optional)">
+            <EuiTextArea
+              value={formData.customSourceConfig}
+              onChange={(e) =>
+                updateField("customSourceConfig", e.target.value)
+              }
+              placeholder='{"key": "value"}'
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "RAY_SOURCE") {
+      return (
+        <>
+          <EuiFormRow
+            label="Reader Type"
+            helpText="Ray Data reader to use for loading data."
+          >
+            <EuiSelect
+              options={RAY_READER_OPTIONS}
+              value={formData.rayReaderType}
+              onChange={(e) => updateField("rayReaderType", e.target.value)}
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Path"
+            isInvalid={!!errors.rayPath}
+            error={errors.rayPath}
+            helpText="File path or directory for file-based readers (s3://, gs://, local)."
+          >
+            <EuiFieldText
+              value={formData.rayPath}
+              onChange={(e) => updateField("rayPath", e.target.value)}
+              isInvalid={!!errors.rayPath}
+              placeholder="s3://bucket/images/"
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Reader Options (JSON, optional)"
+            helpText='e.g. {"dataset_name": "org/dataset", "split": "train"} for HuggingFace'
+          >
+            <EuiTextArea
+              value={formData.rayReaderOptions}
+              onChange={(e) => updateField("rayReaderOptions", e.target.value)}
+              placeholder='{"dataset_name": "org/name", "split": "train"}'
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "POSTGRES_SOURCE") {
+      return (
+        <>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.postgresTable}
+            error={errors.postgresTable}
+            helpText="Table name. Provide either table or query."
+          >
+            <EuiFieldText
+              value={formData.postgresTable}
+              onChange={(e) => updateField("postgresTable", e.target.value)}
+              isInvalid={!!errors.postgresTable}
+              placeholder="public.my_features"
+            />
+          </EuiFormRow>
+          <EuiFormRow label="Query (optional)">
+            <EuiTextArea
+              value={formData.postgresQuery}
+              onChange={(e) => updateField("postgresQuery", e.target.value)}
+              placeholder="SELECT * FROM my_features WHERE ..."
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "MONGODB_SOURCE") {
+      return (
+        <EuiFormRow
+          label="Collection"
+          isInvalid={!!errors.mongodbCollection}
+          error={errors.mongodbCollection}
+          helpText="MongoDB collection name. Connection details are configured in feature_store.yaml."
+        >
+          <EuiFieldText
+            value={formData.mongodbCollection}
+            onChange={(e) => updateField("mongodbCollection", e.target.value)}
+            isInvalid={!!errors.mongodbCollection}
+            placeholder="features_collection"
+          />
+        </EuiFormRow>
+      );
+    }
+
+    if (st === "CLICKHOUSE_SOURCE") {
+      return (
+        <>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.clickhouseTable}
+            error={errors.clickhouseTable}
+            helpText="ClickHouse table name. Provide either table or query."
+          >
+            <EuiFieldText
+              value={formData.clickhouseTable}
+              onChange={(e) => updateField("clickhouseTable", e.target.value)}
+              isInvalid={!!errors.clickhouseTable}
+              placeholder="default.my_features"
+            />
+          </EuiFormRow>
+          <EuiFormRow label="Query (optional)">
+            <EuiTextArea
+              value={formData.clickhouseQuery}
+              onChange={(e) => updateField("clickhouseQuery", e.target.value)}
+              placeholder="SELECT * FROM default.my_features"
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "MSSQL_SOURCE") {
+      return (
+        <>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.mssqlTable}
+            error={errors.mssqlTable}
+          >
+            <EuiFieldText
+              value={formData.mssqlTable}
+              onChange={(e) => updateField("mssqlTable", e.target.value)}
+              isInvalid={!!errors.mssqlTable}
+              placeholder="dbo.my_features"
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Connection String (optional)"
+            helpText="ODBC-style connection string. Can also be set in feature_store.yaml."
+          >
+            <EuiFieldText
+              value={formData.mssqlConnectionStr}
+              onChange={(e) =>
+                updateField("mssqlConnectionStr", e.target.value)
+              }
+              placeholder="mssql+pyodbc://user:pass@host/db" // pragma: allowlist secret
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "ORACLE_SOURCE") {
+      return (
+        <>
+          <EuiFormRow
+            label="Table"
+            isInvalid={!!errors.oracleTable}
+            error={errors.oracleTable}
+          >
+            <EuiFieldText
+              value={formData.oracleTable}
+              onChange={(e) => updateField("oracleTable", e.target.value)}
+              isInvalid={!!errors.oracleTable}
+              placeholder="SCHEMA.MY_FEATURES"
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Connection String (optional)"
+            helpText="Oracle connection string. Can also be set in feature_store.yaml."
+          >
+            <EuiFieldText
+              value={formData.oracleConnectionStr}
+              onChange={(e) =>
+                updateField("oracleConnectionStr", e.target.value)
+              }
+              placeholder="oracle+cx_oracle://user:pass@host:1521/service" // pragma: allowlist secret
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
+    if (st === "COUCHBASE_SOURCE") {
+      return (
+        <>
+          <EuiFlexGroup gutterSize="m">
+            <EuiFlexItem>
+              <EuiFormRow label="Database">
+                <EuiFieldText
+                  value={formData.couchbaseDatabase}
+                  onChange={(e) =>
+                    updateField("couchbaseDatabase", e.target.value)
+                  }
+                  placeholder="Default"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+            <EuiFlexItem>
+              <EuiFormRow label="Scope">
+                <EuiFieldText
+                  value={formData.couchbaseScope}
+                  onChange={(e) =>
+                    updateField("couchbaseScope", e.target.value)
+                  }
+                  placeholder="Default"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+          </EuiFlexGroup>
+          <EuiFormRow
+            label="Collection"
+            isInvalid={!!errors.couchbaseCollection}
+            error={errors.couchbaseCollection}
+            helpText="Provide either a collection or a SQL++ query."
+          >
+            <EuiFieldText
+              value={formData.couchbaseCollection}
+              onChange={(e) =>
+                updateField("couchbaseCollection", e.target.value)
+              }
+              isInvalid={!!errors.couchbaseCollection}
+              placeholder="my_collection"
+            />
+          </EuiFormRow>
+          <EuiFormRow
+            label="Query (optional)"
+            helpText="SQL++ query as an alternative."
+          >
+            <EuiTextArea
+              value={formData.couchbaseQuery}
+              onChange={(e) => updateField("couchbaseQuery", e.target.value)}
+              placeholder="SELECT * FROM `collection`"
+              rows={3}
+            />
+          </EuiFormRow>
+        </>
+      );
+    }
+
     if (
       st === String(feast.core.DataSource.SourceType.REQUEST_SOURCE) ||
       st === String(feast.core.DataSource.SourceType.PUSH_SOURCE)
     ) {
       return (
-        <EuiText size="s" color="subdued">
-          No additional configuration required for this source type.
-        </EuiText>
+        <EuiPanel color="subdued" paddingSize="m">
+          <EuiText size="s" color="subdued">
+            No connection configuration needed. This source type receives data
+            at request time or via push ingestion.
+          </EuiText>
+        </EuiPanel>
       );
     }
 
     return null;
   };
 
+  const sourceTypeName =
+    SOURCE_TYPE_OPTIONS.find((o) => o.value === formData.sourceType)?.text ||
+    "Data Source";
+
   return (
     <FormModal
-      title={isEdit ? "Edit Data Source" : "Create Data Source"}
-      submitLabel={isEdit ? "Update Data Source" : "Create Data Source"}
+      title={
+        isEdit
+          ? `Edit ${sourceTypeName}`
+          : isPreselected
+            ? `New ${sourceTypeName} Connection`
+            : "Create Data Source"
+      }
+      submitLabel={isEdit ? "Update" : "Create Connection"}
       onClose={onClose}
       onSubmit={handleSubmit}
-      width={650}
+      width={720}
       isSubmitting={isSubmitting}
     >
       {submitError && (
         <>
           <EuiCallOut
-            title={
-              isEdit
-                ? "Unable to update data source"
-                : "Unable to create data source"
-            }
+            title="Unable to save data source"
             color="danger"
             iconType="alert"
             size="s"
@@ -474,90 +1085,132 @@ const DataSourceFormModal: React.FC<DataSourceFormModalProps> = ({
         </>
       )}
 
-      <NameDescriptionOwnerFields
-        name={formData.name}
-        description={formData.description}
-        owner={formData.owner}
-        onChangeName={(v) => updateField("name", v)}
-        onChangeDescription={(v) => updateField("description", v)}
-        onChangeOwner={(v) => updateField("owner", v)}
-        nameDisabled={isEdit}
-        nameError={errors.name}
-        nameHelpText="A unique name for this data source."
-        namePlaceholder="e.g. customer_transactions"
-        descriptionPlaceholder="Describe this data source..."
-      />
+      {isPreselected && renderSourceTypeHeader()}
+      {isPreselected && <EuiSpacer size="m" />}
 
-      <EuiFormRow
-        label="Source Type"
-        helpText="The type of underlying storage system for this data source."
-      >
-        <EuiSelect
-          options={SOURCE_TYPE_OPTIONS}
-          value={formData.sourceType}
-          onChange={(e) => {
-            updateField("sourceType", e.target.value);
-            // Clear source-specific errors when type changes
-            setErrors((prev) => {
-              const next = { ...prev };
-              delete next.fileUri;
-              delete next.bigqueryTable;
-              delete next.snowflakeTable;
-              delete next.snowflakeDatabase;
-              delete next.redshiftTable;
-              delete next.redshiftDatabase;
-              delete next.kafkaBootstrapServers;
-              delete next.kafkaTopic;
-              delete next.sparkTable;
-              delete next.timestampField;
-              return next;
-            });
-          }}
-          disabled={isEdit}
+      {/* Section: Identity */}
+      <EuiTitle size="xxs">
+        <h4>Identity</h4>
+      </EuiTitle>
+      <EuiSpacer size="s" />
+
+      <EuiFlexGroup gutterSize="m">
+        <EuiFlexItem>
+          <EuiFormRow
+            label="Name"
+            isInvalid={!!errors.name}
+            error={errors.name}
+            helpText="Unique identifier for this data source."
+          >
+            <EuiFieldText
+              value={formData.name}
+              onChange={(e) => updateField("name", e.target.value)}
+              isInvalid={!!errors.name}
+              disabled={isEdit}
+              placeholder="e.g. customer_transactions"
+            />
+          </EuiFormRow>
+        </EuiFlexItem>
+        <EuiFlexItem>
+          <EuiFormRow label="Owner (optional)">
+            <EuiFieldText
+              value={formData.owner}
+              onChange={(e) => updateField("owner", e.target.value)}
+              placeholder="team@company.com"
+            />
+          </EuiFormRow>
+        </EuiFlexItem>
+      </EuiFlexGroup>
+
+      <EuiFormRow label="Description (optional)">
+        <EuiFieldText
+          value={formData.description}
+          onChange={(e) => updateField("description", e.target.value)}
+          placeholder="Brief description of this data source..."
         />
       </EuiFormRow>
 
+      {!isPreselected && (
+        <>
+          <EuiSpacer size="m" />
+          <EuiFormRow label="Source Type">
+            <EuiSelect
+              options={SOURCE_TYPE_OPTIONS}
+              value={formData.sourceType}
+              onChange={(e) => {
+                updateField("sourceType", e.target.value);
+                setErrors({});
+              }}
+              disabled={isEdit}
+            />
+          </EuiFormRow>
+        </>
+      )}
+
       <EuiSpacer size="m" />
       <EuiHorizontalRule margin="s" />
-      <EuiText size="s">
-        <h4>Source Configuration</h4>
-      </EuiText>
+
+      {/* Section: Connection */}
+      <EuiTitle size="xxs">
+        <h4>Connection Details</h4>
+      </EuiTitle>
       <EuiSpacer size="s" />
 
       {renderSourceTypeFields()}
 
+      {/* Section: Timestamp (for batch sources) */}
       {isBatchSource && (
         <>
           <EuiSpacer size="m" />
-          <EuiFormRow
-            label="Timestamp Field"
-            isInvalid={!!errors.timestampField}
-            error={errors.timestampField}
-            helpText="Column containing the event timestamp. Required for point-in-time correct feature retrieval."
-          >
-            <EuiFieldText
-              value={formData.timestampField}
-              onChange={(e) => updateField("timestampField", e.target.value)}
-              isInvalid={!!errors.timestampField}
-              placeholder="event_timestamp"
-            />
-          </EuiFormRow>
-          <EuiFormRow
-            label="Created Timestamp Column"
-            helpText="Optional: column tracking when the row was written (used for deduplication)."
-          >
-            <EuiFieldText
-              value={formData.createdTimestampColumn}
-              onChange={(e) =>
-                updateField("createdTimestampColumn", e.target.value)
-              }
-              placeholder="created_at"
-            />
-          </EuiFormRow>
+          <EuiHorizontalRule margin="s" />
+          <EuiTitle size="xxs">
+            <h4>Time Configuration</h4>
+          </EuiTitle>
+          <EuiSpacer size="s" />
+
+          <EuiFlexGroup gutterSize="m">
+            <EuiFlexItem>
+              <EuiFormRow
+                label="Timestamp Field"
+                isInvalid={!!errors.timestampField}
+                error={errors.timestampField}
+                helpText="Event timestamp column for point-in-time joins."
+              >
+                <EuiFieldText
+                  value={formData.timestampField}
+                  onChange={(e) =>
+                    updateField("timestampField", e.target.value)
+                  }
+                  isInvalid={!!errors.timestampField}
+                  placeholder="event_timestamp"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+            <EuiFlexItem>
+              <EuiFormRow
+                label="Created Timestamp (optional)"
+                helpText="Used for deduplication."
+              >
+                <EuiFieldText
+                  value={formData.createdTimestampColumn}
+                  onChange={(e) =>
+                    updateField("createdTimestampColumn", e.target.value)
+                  }
+                  placeholder="created_at"
+                />
+              </EuiFormRow>
+            </EuiFlexItem>
+          </EuiFlexGroup>
         </>
       )}
 
+      {/* Section: Tags */}
       <EuiSpacer size="m" />
+      <EuiHorizontalRule margin="s" />
+      <EuiTitle size="xxs">
+        <h4>Tags (optional)</h4>
+      </EuiTitle>
+      <EuiSpacer size="s" />
 
       <TagsEditor
         tags={formData.tags}
diff --git a/ui/src/components/FeatureViewFormModal.tsx b/ui/src/components/FeatureViewFormModal.tsx
index 497defbb057..9e0655d96cf 100644
--- a/ui/src/components/FeatureViewFormModal.tsx
+++ b/ui/src/components/FeatureViewFormModal.tsx
@@ -275,6 +275,28 @@ const FeatureViewFormModal: React.FC<FeatureViewFormModalProps> = ({
         table: dsData.sparkTable,
         path: dsData.sparkPath,
       };
+    } else if (st === String(feast.core.DataSource.SourceType.BATCH_TRINO)) {
+      payload.trino_options = {
+        table: dsData.trinoTable,
+        query: dsData.trinoQuery,
+      };
+    } else if (st === String(feast.core.DataSource.SourceType.BATCH_ATHENA)) {
+      payload.athena_options = {
+        table: dsData.athenaTable,
+        query: dsData.athenaQuery,
+        database: dsData.athenaDatabase,
+        data_source: dsData.athenaDataSource,
+      };
+    } else if (st === String(feast.core.DataSource.SourceType.STREAM_KINESIS)) {
+      payload.kinesis_options = {
+        region: dsData.kinesisRegion,
+        stream_name: dsData.kinesisStreamName,
+      };
+    } else if (st === String(feast.core.DataSource.SourceType.CUSTOM_SOURCE)) {
+      payload.custom_options = {
+        class_name: dsData.customSourceClassName,
+        config: dsData.customSourceConfig,
+      };
     }
 
     applyDataSource.mutate(payload as any, {
diff --git a/ui/src/graphics/data-source-icons.tsx b/ui/src/graphics/data-source-icons.tsx
new file mode 100644
index 00000000000..813c714a560
--- /dev/null
+++ b/ui/src/graphics/data-source-icons.tsx
@@ -0,0 +1,377 @@
+import React from "react";
+
+export const BigQueryIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M14.5 51.5l-2.3-2.3c-5.7-5.7-8.2-14-6.7-21.8 1.5-7.7 6.7-14.2 13.9-17.3 7.1-3.1 15.3-2.5 21.9 1.7l-2.5 3.3c-5.4-3.5-12.2-3.9-18-1.4-5.9 2.5-10.2 7.9-11.4 14.2-1.3 6.4.8 13.2 5.5 17.9l-0.4 5.7z"
+      fill="#4285F4"
+    />
+    <path
+      d="M51.5 49.7L45.8 50l0.3-5.6c4.7-4.7 6.8-11.5 5.5-17.9-1.3-6.4-5.5-11.7-11.4-14.2l1.1-3.7c7.2 3.1 12.4 9.6 13.9 17.3 1.5 7.8-1 16.1-6.7 21.8l3 2z"
+      fill="#4285F4"
+    />
+    <path
+      d="M32 22c5.5 0 10 4.5 10 10s-4.5 10-10 10-10-4.5-10-10 4.5-10 10-10zm0 4c-3.3 0-6 2.7-6 6s2.7 6 6 6 6-2.7 6-6-2.7-6-6-6z"
+      fill="#4285F4"
+    />
+    <path d="M32 34h12v4H32z" fill="#4285F4" />
+  </svg>
+);
+
+export const SnowflakeIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M32 4v56M11.1 20l41.8 24M11.1 44l41.8-24"
+      stroke="#29B5E8"
+      strokeWidth="4"
+      strokeLinecap="round"
+    />
+    <path
+      d="M32 4l-5 5M32 4l5 5M32 60l-5-5M32 60l5-5M11.1 20l1.5 6.5M11.1 20l6.5-1.5M52.9 44l-1.5-6.5M52.9 44l-6.5 1.5M11.1 44l1.5-6.5M11.1 44l6.5 1.5M52.9 20l-1.5 6.5M52.9 20l-6.5-1.5"
+      stroke="#29B5E8"
+      strokeWidth="3"
+      strokeLinecap="round"
+    />
+  </svg>
+);
+
+export const RedshiftIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path d="M32 8L12 20v24l20 12 20-12V20L32 8z" fill="#205B97" />
+    <path d="M32 8L12 20l20 12 20-12L32 8z" fill="#5294CF" />
+    <path d="M32 32v24l20-12V20L32 32z" fill="#2E73B8" />
+    <path d="M32 32L12 20v24l20 12V32z" fill="#205B97" />
+  </svg>
+);
+
+export const KafkaIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <circle cx="32" cy="20" r="6" fill="#231F20" />
+    <circle cx="20" cy="38" r="5" fill="#231F20" />
+    <circle cx="44" cy="38" r="5" fill="#231F20" />
+    <circle cx="32" cy="50" r="4" fill="#231F20" />
+    <circle cx="18" cy="26" r="4" fill="#231F20" />
+    <circle cx="46" cy="26" r="4" fill="#231F20" />
+    <line x1="32" y1="26" x2="32" y2="46" stroke="#231F20" strokeWidth="2" />
+    <line x1="27" y1="22" x2="22" y2="26" stroke="#231F20" strokeWidth="2" />
+    <line x1="37" y1="22" x2="42" y2="26" stroke="#231F20" strokeWidth="2" />
+    <line x1="20" y1="30" x2="21" y2="33" stroke="#231F20" strokeWidth="2" />
+    <line x1="44" y1="30" x2="43" y2="33" stroke="#231F20" strokeWidth="2" />
+    <line x1="24" y1="40" x2="28" y2="48" stroke="#231F20" strokeWidth="2" />
+    <line x1="40" y1="40" x2="36" y2="48" stroke="#231F20" strokeWidth="2" />
+  </svg>
+);
+
+export const SparkIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M32 6c-2 8-8 14-14 18 6 2 12 8 14 16 2-8 8-14 14-16-6-4-12-10-14-18z"
+      fill="#E25A1C"
+    />
+    <path
+      d="M18 36c-1.5 5-5 9-9 12 4 1.5 8 5.5 9 10 1.5-5 5-9 9-10-4-3-7.5-7-9-12z"
+      fill="#E25A1C"
+      opacity="0.7"
+    />
+    <path
+      d="M46 36c-1 4-4 7-7 9 3 1 6 4.5 7 8 1-4 4-7 7-8-3-2-6-5-7-9z"
+      fill="#E25A1C"
+      opacity="0.5"
+    />
+  </svg>
+);
+
+export const FileIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M16 8h22l12 12v36a2 2 0 01-2 2H16a2 2 0 01-2-2V10a2 2 0 012-2z"
+      fill="#4CAF50"
+      opacity="0.9"
+    />
+    <path d="M38 8l12 12H40a2 2 0 01-2-2V8z" fill="#388E3C" />
+    <rect
+      x="20"
+      y="28"
+      width="24"
+      height="2"
+      rx="1"
+      fill="white"
+      opacity="0.7"
+    />
+    <rect
+      x="20"
+      y="34"
+      width="20"
+      height="2"
+      rx="1"
+      fill="white"
+      opacity="0.7"
+    />
+    <rect
+      x="20"
+      y="40"
+      width="22"
+      height="2"
+      rx="1"
+      fill="white"
+      opacity="0.7"
+    />
+    <rect
+      x="20"
+      y="46"
+      width="16"
+      height="2"
+      rx="1"
+      fill="white"
+      opacity="0.7"
+    />
+  </svg>
+);
+
+export const RequestSourceIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect
+      x="10"
+      y="14"
+      width="44"
+      height="36"
+      rx="4"
+      fill="#7B61FF"
+      opacity="0.9"
+    />
+    <path d="M20 28l6 4-6 4V28z" fill="white" />
+    <rect
+      x="30"
+      y="30"
+      width="14"
+      height="4"
+      rx="2"
+      fill="white"
+      opacity="0.7"
+    />
+    <rect
+      x="14"
+      y="18"
+      width="8"
+      height="3"
+      rx="1.5"
+      fill="white"
+      opacity="0.5"
+    />
+    <circle cx="48" cy="19.5" r="2" fill="#4CAF50" />
+  </svg>
+);
+
+export const PushSourceIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect
+      x="14"
+      y="16"
+      width="36"
+      height="32"
+      rx="3"
+      fill="#FF6B35"
+      opacity="0.9"
+    />
+    <path
+      d="M32 24v16M26 34l6 6 6-6"
+      stroke="white"
+      strokeWidth="3"
+      strokeLinecap="round"
+      strokeLinejoin="round"
+    />
+    <rect x="22" y="12" width="20" height="4" rx="2" fill="#FF6B35" />
+  </svg>
+);
+
+export const KinesisIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path d="M16 12h6l10 20-10 20h-6l10-20L16 12z" fill="#FF9900" />
+    <path
+      d="M28 12h6l10 20-10 20h-6l10-20L28 12z"
+      fill="#FF9900"
+      opacity="0.7"
+    />
+    <path
+      d="M40 12h6l10 20-10 20h-6l10-20L40 12z"
+      fill="#FF9900"
+      opacity="0.5"
+    />
+  </svg>
+);
+
+export const TrinoIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <circle cx="32" cy="32" r="22" fill="#DD00A1" opacity="0.1" />
+    <path
+      d="M32 10c-12.15 0-22 9.85-22 22s9.85 22 22 22 22-9.85 22-22-9.85-22-22-22zm0 4c9.94 0 18 8.06 18 18s-8.06 18-18 18-18-8.06-18-18 8.06-18 18-18z"
+      fill="#DD00A1"
+    />
+    <path d="M26 26h4v12h-4zM34 22h4v16h-4z" fill="#DD00A1" />
+    <circle cx="28" cy="22" r="2.5" fill="#DD00A1" />
+    <circle cx="36" cy="42" r="2.5" fill="#DD00A1" />
+  </svg>
+);
+
+export const AthenaIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M32 8L14 56h6l4-12h16l4 12h6L32 8zm0 12l6 20H26l6-20z"
+      fill="#8C4FFF"
+    />
+    <rect
+      x="20"
+      y="48"
+      width="24"
+      height="3"
+      rx="1.5"
+      fill="#8C4FFF"
+      opacity="0.5"
+    />
+  </svg>
+);
+
+export const CustomSourceIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect
+      x="12"
+      y="12"
+      width="40"
+      height="40"
+      rx="6"
+      fill="#607D8B"
+      opacity="0.15"
+    />
+    <path
+      d="M24 28l-6 4 6 4M40 28l6 4-6 4M35 22l-6 20"
+      stroke="#607D8B"
+      strokeWidth="3"
+      strokeLinecap="round"
+      strokeLinejoin="round"
+    />
+  </svg>
+);
+
+export const RayIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <circle cx="32" cy="32" r="20" fill="#00A2E8" opacity="0.12" />
+    <path
+      d="M32 12v40M12 32h40M18 18l28 28M46 18L18 46"
+      stroke="#00A2E8"
+      strokeWidth="3"
+      strokeLinecap="round"
+    />
+    <circle cx="32" cy="32" r="5" fill="#00A2E8" />
+  </svg>
+);
+
+export const PostgresIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M32 8C20 8 14 14 14 22c0 6 4 10 4 14v8c0 6 6 12 14 12s14-6 14-12v-8c0-4 4-8 4-14 0-8-6-14-18-14z"
+      fill="#336791"
+      opacity="0.9"
+    />
+    <ellipse cx="32" cy="22" rx="12" ry="8" fill="#336791" />
+    <ellipse cx="32" cy="22" rx="10" ry="6" fill="white" opacity="0.3" />
+    <path
+      d="M26 30v12c0 4 3 7 6 7s6-3 6-7V30"
+      stroke="white"
+      strokeWidth="2"
+      opacity="0.5"
+    />
+  </svg>
+);
+
+export const MongoDBIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <path
+      d="M32 6c-1.5 8-4 14-4 22 0 10 2 18 4 30 2-12 4-20 4-30 0-8-2.5-14-4-22z"
+      fill="#00ED64"
+    />
+    <path
+      d="M32 6c-6 10-16 16-16 28 0 10 8 18 16 24"
+      fill="#00684A"
+      opacity="0.6"
+    />
+    <path
+      d="M32 6c6 10 16 16 16 28 0 10-8 18-16 24"
+      fill="#00ED64"
+      opacity="0.6"
+    />
+  </svg>
+);
+
+export const SqlServerIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect
+      x="12"
+      y="16"
+      width="40"
+      height="32"
+      rx="4"
+      fill="#CC2927"
+      opacity="0.9"
+    />
+    <path d="M20 28h8v2h-8zM20 34h12v2H20z" fill="white" opacity="0.7" />
+    <ellipse cx="42" cy="26" rx="6" ry="4" fill="white" opacity="0.5" />
+    <path
+      d="M36 26v12c0 2 3 4 6 4s6-2 6-4V26"
+      stroke="white"
+      strokeWidth="1.5"
+      opacity="0.5"
+    />
+  </svg>
+);
+
+export const OracleIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect
+      x="10"
+      y="20"
+      width="44"
+      height="24"
+      rx="12"
+      fill="#F80000"
+      opacity="0.9"
+    />
+    <rect
+      x="16"
+      y="26"
+      width="32"
+      height="12"
+      rx="6"
+      fill="white"
+      opacity="0.3"
+    />
+    <text
+      x="32"
+      y="36"
+      textAnchor="middle"
+      fill="white"
+      fontSize="10"
+      fontWeight="bold"
+    >
+      ORA
+    </text>
+  </svg>
+);
+
+export const CouchbaseIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <circle cx="32" cy="32" r="22" fill="#EA2328" opacity="0.15" />
+    <circle cx="32" cy="32" r="16" fill="#EA2328" opacity="0.3" />
+    <circle cx="32" cy="32" r="8" fill="#EA2328" />
+    <path d="M32 24v16M24 32h16" stroke="white" strokeWidth="2" />
+  </svg>
+);
+
+export const ClickHouseIcon = (props: React.SVGProps<SVGSVGElement>) => (
+  <svg {...props} viewBox="0 0 64 64" fill="none">
+    <rect x="14" y="12" width="6" height="40" fill="#FFCC00" />
+    <rect x="22" y="12" width="6" height="40" fill="#FFCC00" />
+    <rect x="30" y="12" width="6" height="40" fill="#FFCC00" />
+    <rect x="38" y="12" width="6" height="40" fill="#FFCC00" />
+    <rect x="46" y="20" width="6" height="24" rx="3" fill="#FFCC00" />
+  </svg>
+);
diff --git a/ui/src/pages/data-sources/DataSourceCatalog.tsx b/ui/src/pages/data-sources/DataSourceCatalog.tsx
new file mode 100644
index 00000000000..fef72186bb0
--- /dev/null
+++ b/ui/src/pages/data-sources/DataSourceCatalog.tsx
@@ -0,0 +1,369 @@
+import React, { useState } from "react";
+import {
+  EuiFlexGroup,
+  EuiFlexItem,
+  EuiPanel,
+  EuiText,
+  EuiTitle,
+  EuiSpacer,
+  EuiButton,
+  EuiBadge,
+} from "@elastic/eui";
+import { feast } from "../../protos";
+import {
+  BigQueryIcon,
+  SnowflakeIcon,
+  RedshiftIcon,
+  KafkaIcon,
+  SparkIcon,
+  FileIcon,
+  RequestSourceIcon,
+  PushSourceIcon,
+  KinesisIcon,
+  TrinoIcon,
+  AthenaIcon,
+  CustomSourceIcon,
+  RayIcon,
+  PostgresIcon,
+  MongoDBIcon,
+  SqlServerIcon,
+  OracleIcon,
+  CouchbaseIcon,
+  ClickHouseIcon,
+} from "../../graphics/data-source-icons";
+
+interface DataSourceTypeInfo {
+  id: string;
+  sourceType: string;
+  name: string;
+  description: string;
+  icon: React.FC<React.SVGProps<SVGSVGElement>>;
+  category: "batch" | "stream" | "on-demand";
+  color: string;
+  contrib?: boolean;
+}
+
+const DATA_SOURCE_TYPES: DataSourceTypeInfo[] = [
+  {
+    id: "bigquery",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_BIGQUERY),
+    name: "BigQuery",
+    description:
+      "Google Cloud's serverless data warehouse. Ideal for large-scale analytics and ML feature computation.",
+    icon: BigQueryIcon,
+    category: "batch",
+    color: "#4285F4",
+  },
+  {
+    id: "snowflake",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_SNOWFLAKE),
+    name: "Snowflake",
+    description:
+      "Cloud-native data platform with elastic scaling. Connect to your Snowflake tables for feature engineering.",
+    icon: SnowflakeIcon,
+    category: "batch",
+    color: "#29B5E8",
+  },
+  {
+    id: "redshift",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_REDSHIFT),
+    name: "Redshift",
+    description:
+      "AWS fully managed data warehouse. Pull features from your Redshift clusters with fast parallel queries.",
+    icon: RedshiftIcon,
+    category: "batch",
+    color: "#205B97",
+  },
+  {
+    id: "spark",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_SPARK),
+    name: "Spark",
+    description:
+      "Apache Spark data source for distributed processing. Access tables via Spark catalog or direct file paths.",
+    icon: SparkIcon,
+    category: "batch",
+    color: "#E25A1C",
+  },
+  {
+    id: "file",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_FILE),
+    name: "File (Parquet / CSV)",
+    description:
+      "Read features from Parquet or CSV files stored in S3, GCS, HDFS, or local filesystem.",
+    icon: FileIcon,
+    category: "batch",
+    color: "#4CAF50",
+  },
+  {
+    id: "trino",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_TRINO),
+    name: "Trino",
+    description:
+      "Distributed SQL query engine for big data analytics. Query data across heterogeneous sources via Trino catalog.",
+    icon: TrinoIcon,
+    category: "batch",
+    color: "#DD00A1",
+  },
+  {
+    id: "athena",
+    sourceType: String(feast.core.DataSource.SourceType.BATCH_ATHENA),
+    name: "AWS Athena",
+    description:
+      "Serverless interactive query service on AWS. Run SQL queries directly against data in S3 without infrastructure.",
+    icon: AthenaIcon,
+    category: "batch",
+    color: "#8C4FFF",
+  },
+  {
+    id: "kafka",
+    sourceType: String(feast.core.DataSource.SourceType.STREAM_KAFKA),
+    name: "Kafka",
+    description:
+      "Real-time event streaming platform. Ingest features from Kafka topics for low-latency serving.",
+    icon: KafkaIcon,
+    category: "stream",
+    color: "#231F20",
+  },
+  {
+    id: "kinesis",
+    sourceType: String(feast.core.DataSource.SourceType.STREAM_KINESIS),
+    name: "AWS Kinesis",
+    description:
+      "Managed real-time data streaming on AWS. Capture and process streaming data at scale for real-time features.",
+    icon: KinesisIcon,
+    category: "stream",
+    color: "#FF9900",
+  },
+  {
+    id: "request-source",
+    sourceType: String(feast.core.DataSource.SourceType.REQUEST_SOURCE),
+    name: "Request Source",
+    description:
+      "Features provided at request time by the caller. No external storage needed — values come from the client.",
+    icon: RequestSourceIcon,
+    category: "on-demand",
+    color: "#7B61FF",
+  },
+  {
+    id: "push-source",
+    sourceType: String(feast.core.DataSource.SourceType.PUSH_SOURCE),
+    name: "Push Source",
+    description:
+      "Push-based ingestion source. Clients push feature values directly to the online/offline store.",
+    icon: PushSourceIcon,
+    category: "on-demand",
+    color: "#FF6B35",
+  },
+  {
+    id: "ray",
+    sourceType: "RAY_SOURCE",
+    name: "Ray",
+    description:
+      "Multi-format data source powered by Ray. Read images, HuggingFace datasets, Parquet, CSV, MongoDB, and more via Ray Data.",
+    icon: RayIcon,
+    category: "batch",
+    color: "#00A2E8",
+    contrib: true,
+  },
+  {
+    id: "postgres",
+    sourceType: "POSTGRES_SOURCE",
+    name: "PostgreSQL",
+    description:
+      "Open-source relational database. Query feature data from PostgreSQL tables with full SQL support.",
+    icon: PostgresIcon,
+    category: "batch",
+    color: "#336791",
+    contrib: true,
+  },
+  {
+    id: "mongodb",
+    sourceType: "MONGODB_SOURCE",
+    name: "MongoDB",
+    description:
+      "Document-oriented NoSQL database. Access feature data stored in MongoDB collections.",
+    icon: MongoDBIcon,
+    category: "batch",
+    color: "#00684A",
+    contrib: true,
+  },
+  {
+    id: "clickhouse",
+    sourceType: "CLICKHOUSE_SOURCE",
+    name: "ClickHouse",
+    description:
+      "Column-oriented OLAP database for real-time analytics. High-performance queries for feature retrieval.",
+    icon: ClickHouseIcon,
+    category: "batch",
+    color: "#FFCC00",
+    contrib: true,
+  },
+  {
+    id: "mssql",
+    sourceType: "MSSQL_SOURCE",
+    name: "SQL Server",
+    description:
+      "Microsoft SQL Server data source. Connect to MSSQL databases for enterprise feature data.",
+    icon: SqlServerIcon,
+    category: "batch",
+    color: "#CC2927",
+    contrib: true,
+  },
+  {
+    id: "oracle",
+    sourceType: "ORACLE_SOURCE",
+    name: "Oracle",
+    description:
+      "Oracle Database data source. Pull features from Oracle tables and views for enterprise workloads.",
+    icon: OracleIcon,
+    category: "batch",
+    color: "#F80000",
+    contrib: true,
+  },
+  {
+    id: "couchbase",
+    sourceType: "COUCHBASE_SOURCE",
+    name: "Couchbase",
+    description:
+      "Couchbase Columnar analytics source. Run SQL++ queries across distributed data in Couchbase.",
+    icon: CouchbaseIcon,
+    category: "batch",
+    color: "#EA2328",
+    contrib: true,
+  },
+  {
+    id: "custom-source",
+    sourceType: String(feast.core.DataSource.SourceType.CUSTOM_SOURCE),
+    name: "Custom Source",
+    description:
+      "Plugin-based data source for custom integrations. Extend Feast with your own data source implementation.",
+    icon: CustomSourceIcon,
+    category: "batch",
+    color: "#607D8B",
+  },
+];
+
+const CATEGORY_LABELS: Record<string, { label: string; color: string }> = {
+  batch: { label: "Batch", color: "primary" },
+  stream: { label: "Streaming", color: "accent" },
+  "on-demand": { label: "On-Demand", color: "warning" },
+};
+
+interface DataSourceCatalogProps {
+  onSelectType: (sourceType: string) => void;
+}
+
+const SourceCard: React.FC<{
+  dsType: DataSourceTypeInfo;
+  isHovered: boolean;
+  onHover: (id: string | null) => void;
+  onSelect: (sourceType: string) => void;
+}> = ({ dsType, isHovered, onHover, onSelect }) => {
+  const categoryInfo = CATEGORY_LABELS[dsType.category];
+
+  return (
+    <EuiFlexItem grow={false} style={{ width: 320, minWidth: 280 }}>
+      <EuiPanel
+        hasBorder
+        hasShadow={isHovered}
+        paddingSize="l"
+        onMouseEnter={() => onHover(dsType.id)}
+        onMouseLeave={() => onHover(null)}
+        style={{
+          height: "100%",
+          display: "flex",
+          flexDirection: "column",
+          transition: "all 0.2s ease",
+          transform: isHovered ? "translateY(-2px)" : "none",
+          borderTop: `3px solid ${dsType.color}`,
+          cursor: "pointer",
+        }}
+        onClick={() => onSelect(dsType.sourceType)}
+      >
+        <EuiFlexGroup alignItems="center" gutterSize="m" responsive={false}>
+          <EuiFlexItem grow={false}>
+            <div
+              style={{
+                width: 48,
+                height: 48,
+                display: "flex",
+                alignItems: "center",
+                justifyContent: "center",
+                borderRadius: 8,
+                backgroundColor: `${dsType.color}10`,
+              }}
+            >
+              <dsType.icon width={32} height={32} />
+            </div>
+          </EuiFlexItem>
+          <EuiFlexItem>
+            <EuiTitle size="xs">
+              <h3 style={{ margin: 0 }}>{dsType.name}</h3>
+            </EuiTitle>
+          </EuiFlexItem>
+          <EuiFlexItem grow={false}>
+            <EuiBadge color={categoryInfo.color as any}>
+              {categoryInfo.label}
+            </EuiBadge>
+          </EuiFlexItem>
+        </EuiFlexGroup>
+
+        <EuiSpacer size="m" />
+
+        <EuiText size="s" color="subdued" style={{ flex: 1 }}>
+          <p style={{ margin: 0 }}>{dsType.description}</p>
+        </EuiText>
+
+        <EuiSpacer size="m" />
+
+        <EuiButton
+          fullWidth
+          color="primary"
+          fill={isHovered}
+          onClick={(e: React.MouseEvent) => {
+            e.stopPropagation();
+            onSelect(dsType.sourceType);
+          }}
+          iconType="plusInCircle"
+          size="s"
+        >
+          Create Connection
+        </EuiButton>
+      </EuiPanel>
+    </EuiFlexItem>
+  );
+};
+
+const DataSourceCatalog: React.FC<DataSourceCatalogProps> = ({
+  onSelectType,
+}) => {
+  const [hoveredId, setHoveredId] = useState<string | null>(null);
+
+  return (
+    <div>
+      <EuiText>
+        <p style={{ color: "#69707D", maxWidth: 640 }}>
+          Choose a data source type to create a new connection. Each type has
+          its own configuration tailored to the underlying storage system.
+        </p>
+      </EuiText>
+      <EuiSpacer size="l" />
+
+      <EuiFlexGroup wrap responsive gutterSize="l">
+        {DATA_SOURCE_TYPES.map((dsType) => (
+          <SourceCard
+            key={dsType.id}
+            dsType={dsType}
+            isHovered={hoveredId === dsType.id}
+            onHover={setHoveredId}
+            onSelect={onSelectType}
+          />
+        ))}
+      </EuiFlexGroup>
+    </div>
+  );
+};
+
+export default DataSourceCatalog;
+export { DATA_SOURCE_TYPES };
+export type { DataSourceTypeInfo };
diff --git a/ui/src/pages/data-sources/DataSourceOverviewTab.tsx b/ui/src/pages/data-sources/DataSourceOverviewTab.tsx
index 6ee71f9a65c..42a8d2536f0 100644
--- a/ui/src/pages/data-sources/DataSourceOverviewTab.tsx
+++ b/ui/src/pages/data-sources/DataSourceOverviewTab.tsx
@@ -69,6 +69,38 @@ const buildEditFormData = (ds: any): DataSourceFormData => {
     kafkaTopic: spec.kafkaOptions?.topic || ds.kafkaOptions?.topic || "",
     sparkTable: spec.sparkOptions?.table || ds.sparkOptions?.table || "",
     sparkPath: spec.sparkOptions?.path || ds.sparkOptions?.path || "",
+    kinesisRegion:
+      spec.kinesisOptions?.region || ds.kinesisOptions?.region || "",
+    kinesisStreamName:
+      spec.kinesisOptions?.streamName || ds.kinesisOptions?.streamName || "",
+    trinoTable: spec.trinoOptions?.table || ds.trinoOptions?.table || "",
+    trinoQuery: spec.trinoOptions?.query || ds.trinoOptions?.query || "",
+    athenaTable: spec.athenaOptions?.table || ds.athenaOptions?.table || "",
+    athenaQuery: spec.athenaOptions?.query || ds.athenaOptions?.query || "",
+    athenaDatabase:
+      spec.athenaOptions?.database || ds.athenaOptions?.database || "",
+    athenaDataSource:
+      spec.athenaOptions?.dataSource || ds.athenaOptions?.dataSource || "",
+    customSourceClassName:
+      spec.customOptions?.className || ds.customOptions?.className || "",
+    customSourceConfig:
+      spec.customOptions?.config || ds.customOptions?.config || "",
+    rayReaderType: "",
+    rayPath: "",
+    rayReaderOptions: "",
+    postgresTable: "",
+    postgresQuery: "",
+    mongodbCollection: "",
+    clickhouseTable: "",
+    clickhouseQuery: "",
+    mssqlTable: "",
+    mssqlConnectionStr: "",
+    oracleTable: "",
+    oracleConnectionStr: "",
+    couchbaseDatabase: "",
+    couchbaseScope: "",
+    couchbaseCollection: "",
+    couchbaseQuery: "",
   };
 };
 
@@ -116,6 +148,28 @@ const formDataToPayload = (formData: DataSourceFormData, project: string) => {
       table: formData.sparkTable,
       path: formData.sparkPath,
     };
+  } else if (st === String(feast.core.DataSource.SourceType.BATCH_TRINO)) {
+    payload.trino_options = {
+      table: formData.trinoTable,
+      query: formData.trinoQuery,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.BATCH_ATHENA)) {
+    payload.athena_options = {
+      table: formData.athenaTable,
+      query: formData.athenaQuery,
+      database: formData.athenaDatabase,
+      data_source: formData.athenaDataSource,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.STREAM_KINESIS)) {
+    payload.kinesis_options = {
+      region: formData.kinesisRegion,
+      stream_name: formData.kinesisStreamName,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.CUSTOM_SOURCE)) {
+    payload.custom_options = {
+      class_name: formData.customSourceClassName,
+      config: formData.customSourceConfig,
+    };
   }
 
   return payload;
diff --git a/ui/src/pages/data-sources/Index.tsx b/ui/src/pages/data-sources/Index.tsx
index cf44de1770f..0a6a0d7148c 100644
--- a/ui/src/pages/data-sources/Index.tsx
+++ b/ui/src/pages/data-sources/Index.tsx
@@ -1,4 +1,4 @@
-import React, { useState } from "react";
+import React, { useMemo, useState } from "react";
 import { useParams } from "react-router-dom";
 
 import {
@@ -14,8 +14,8 @@ import {
 } from "@elastic/eui";
 
 import DatasourcesListingTable from "./DataSourcesListingTable";
+import DataSourceCatalog from "./DataSourceCatalog";
 import { useDocumentTitle } from "../../hooks/useDocumentTitle";
-import DataSourceIndexEmptyState from "./DataSourceIndexEmptyState";
 import { DataSourceIcon } from "../../graphics/DataSourceIcon";
 import { useSearchQuery } from "../../hooks/useSearchInputWithTags";
 import { feast } from "../../protos";
@@ -95,6 +95,28 @@ const formDataToPayload = (formData: DataSourceFormData, project: string) => {
       table: formData.sparkTable,
       path: formData.sparkPath,
     };
+  } else if (st === String(feast.core.DataSource.SourceType.BATCH_TRINO)) {
+    payload.trino_options = {
+      table: formData.trinoTable,
+      query: formData.trinoQuery,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.BATCH_ATHENA)) {
+    payload.athena_options = {
+      table: formData.athenaTable,
+      query: formData.athenaQuery,
+      database: formData.athenaDatabase,
+      data_source: formData.athenaDataSource,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.STREAM_KINESIS)) {
+    payload.kinesis_options = {
+      region: formData.kinesisRegion,
+      stream_name: formData.kinesisStreamName,
+    };
+  } else if (st === String(feast.core.DataSource.SourceType.CUSTOM_SOURCE)) {
+    payload.custom_options = {
+      class_name: formData.customSourceClassName,
+      config: formData.customSourceConfig,
+    };
   }
 
   return payload;
@@ -105,7 +127,11 @@ const Index = () => {
   const { isLoading, isSuccess, isError, data } = useLoadDatasources();
   const isAllProjects = projectName === "all";
 
+  const [showCatalog, setShowCatalog] = useState(false);
   const [isModalOpen, setIsModalOpen] = useState(false);
+  const [preselectedSourceType, setPreselectedSourceType] = useState<
+    string | null
+  >(null);
   const [successMessage, setSuccessMessage] = useState<string | null>(null);
   const [errorMessage, setErrorMessage] = useState<string | null>(null);
   const applyDataSource = useApplyDataSource();
@@ -116,11 +142,73 @@ const Index = () => {
 
   const filterResult = data ? filterFn(data, searchTokens) : data;
 
+  const hasExistingSources = isSuccess && data && data.length > 0;
+  const isEmpty = isSuccess && (!data || data.length === 0);
+
+  const modalInitialData = useMemo(() => {
+    if (!preselectedSourceType) return undefined;
+    return {
+      name: "",
+      description: "",
+      owner: "",
+      sourceType: preselectedSourceType,
+      timestampField: "",
+      createdTimestampColumn: "",
+      tags: [] as { key: string; value: string }[],
+      fileUri: "",
+      bigqueryTable: "",
+      bigqueryQuery: "",
+      snowflakeTable: "",
+      snowflakeDatabase: "",
+      snowflakeSchema: "",
+      redshiftTable: "",
+      redshiftDatabase: "",
+      redshiftSchema: "",
+      kafkaBootstrapServers: "",
+      kafkaTopic: "",
+      sparkTable: "",
+      sparkPath: "",
+      kinesisRegion: "",
+      kinesisStreamName: "",
+      trinoTable: "",
+      trinoQuery: "",
+      athenaTable: "",
+      athenaQuery: "",
+      athenaDatabase: "",
+      athenaDataSource: "",
+      customSourceClassName: "",
+      customSourceConfig: "",
+      rayReaderType: "parquet",
+      rayPath: "",
+      rayReaderOptions: "",
+      postgresTable: "",
+      postgresQuery: "",
+      mongodbCollection: "",
+      clickhouseTable: "",
+      clickhouseQuery: "",
+      mssqlTable: "",
+      mssqlConnectionStr: "",
+      oracleTable: "",
+      oracleConnectionStr: "",
+      couchbaseDatabase: "",
+      couchbaseScope: "",
+      couchbaseCollection: "",
+      couchbaseQuery: "",
+    };
+  }, [preselectedSourceType]);
+
+  const handleSelectType = (sourceType: string) => {
+    setPreselectedSourceType(sourceType);
+    setIsModalOpen(true);
+  };
+
   const handleCreateSubmit = (formData: DataSourceFormData) => {
     const payload = formDataToPayload(formData, projectName || "");
     applyDataSource.mutate(payload as any, {
       onSuccess: () => {
         setIsModalOpen(false);
+        setPreselectedSourceType(null);
+        setShowCatalog(false);
         setErrorMessage(null);
         setSuccessMessage(
           `Data source "${formData.name}" created successfully.`,
@@ -128,7 +216,6 @@ const Index = () => {
         setTimeout(() => setSuccessMessage(null), 5000);
       },
       onError: (err: unknown) => {
-        // Error shown inside the modal via submitError prop
         const message =
           err instanceof Error ? err.message : "An unexpected error occurred.";
         setErrorMessage(message);
@@ -143,13 +230,13 @@ const Index = () => {
         iconType={DataSourceIcon}
         pageTitle="Data Sources"
         rightSideItems={[
-          ...(isAllProjects
+          ...(isAllProjects || showCatalog
             ? []
             : [
                 <EuiButton
                   fill
                   iconType="plus"
-                  onClick={() => setIsModalOpen(true)}
+                  onClick={() => setShowCatalog(true)}
                   key="create"
                 >
                   Create Data Source
@@ -175,7 +262,7 @@ const Index = () => {
             <EuiSpacer size="m" />
           </>
         )}
-        {errorMessage && (
+        {errorMessage && !isModalOpen && (
           <>
             <EuiCallOut
               title={errorMessage}
@@ -186,32 +273,73 @@ const Index = () => {
             <EuiSpacer size="m" />
           </>
         )}
-        {isLoading && (
-          <p>
-            <EuiLoadingSpinner size="m" /> Loading
-          </p>
-        )}
-        {isError && <p>We encountered an error while loading.</p>}
-        {isSuccess && !data && <DataSourceIndexEmptyState />}
-        {isSuccess && data && data.length > 0 && filterResult && (
-          <React.Fragment>
-            <EuiFlexGroup>
-              <EuiFlexItem grow={2}>
-                <EuiTitle size="xs">
-                  <h2>Search</h2>
+
+        {showCatalog && !isAllProjects && (
+          <>
+            <EuiFlexGroup alignItems="center" justifyContent="spaceBetween">
+              <EuiFlexItem grow={false}>
+                <EuiTitle size="s">
+                  <h2>Select a Data Source Type</h2>
                 </EuiTitle>
-                <EuiFieldSearch
-                  value={searchString}
-                  fullWidth={true}
-                  onChange={(e) => {
-                    setSearchString(e.target.value);
-                  }}
-                />
               </EuiFlexItem>
+              {hasExistingSources && (
+                <EuiFlexItem grow={false}>
+                  <EuiButton
+                    onClick={() => setShowCatalog(false)}
+                    iconType="arrowLeft"
+                    size="s"
+                  >
+                    Back to Data Sources
+                  </EuiButton>
+                </EuiFlexItem>
+              )}
             </EuiFlexGroup>
             <EuiSpacer size="m" />
-            <DatasourcesListingTable dataSources={filterResult} />
-          </React.Fragment>
+            <DataSourceCatalog onSelectType={handleSelectType} />
+          </>
+        )}
+
+        {!showCatalog && (
+          <>
+            {isLoading && (
+              <p>
+                <EuiLoadingSpinner size="m" /> Loading
+              </p>
+            )}
+            {isError && <p>We encountered an error while loading.</p>}
+            {isEmpty && !isAllProjects && (
+              <>
+                <EuiTitle size="s">
+                  <h2>No data sources yet — create your first connection</h2>
+                </EuiTitle>
+                <EuiSpacer size="l" />
+                <DataSourceCatalog onSelectType={handleSelectType} />
+              </>
+            )}
+            {isEmpty && isAllProjects && (
+              <p>No data sources found across projects.</p>
+            )}
+            {hasExistingSources && filterResult && (
+              <React.Fragment>
+                <EuiFlexGroup>
+                  <EuiFlexItem grow={2}>
+                    <EuiTitle size="xs">
+                      <h2>Search</h2>
+                    </EuiTitle>
+                    <EuiFieldSearch
+                      value={searchString}
+                      fullWidth={true}
+                      onChange={(e) => {
+                        setSearchString(e.target.value);
+                      }}
+                    />
+                  </EuiFlexItem>
+                </EuiFlexGroup>
+                <EuiSpacer size="m" />
+                <DatasourcesListingTable dataSources={filterResult} />
+              </React.Fragment>
+            )}
+          </>
         )}
       </EuiPageTemplate.Section>
 
@@ -219,11 +347,13 @@ const Index = () => {
         <DataSourceFormModal
           onClose={() => {
             setIsModalOpen(false);
+            setPreselectedSourceType(null);
             setErrorMessage(null);
           }}
           onSubmit={handleCreateSubmit}
           isSubmitting={applyDataSource.isLoading}
           submitError={errorMessage}
+          initialData={modalInitialData}
         />
       )}
     </EuiPageTemplate>