Kuscia添加数据源类型，例如：Oracle、hive、Doris等多种类型数据库

[Trae-Young]

Issue Type

Api Usage

Search for existing issues similar to yours

Yes

Kuscia Version

kuscia 0.12.0b0

Link to Relevant Documentation

https://www.secretflow.org.cn/zh-CN/docs/kuscia/v0.12.0b0/reference/apis/domaindatasource_cn

Question Details

了解到domainDatasource是管理kuscia的数据源。支持localfs、oss、MySQL类型的数据源。
业务场景一般是通过secretpad进行注册数据、创建项目、进行数据分析、数据建模、模型训练等。
现在希望能对kuscia的数据源类型进行二开，添加类似Oracle、hive、Doris类型的数据源。在改造的过程中，是不是kuscia和secretpad都需要进行深度改造？
大概是那些地方需要进行改造？

[wangzul]

数据源管理和实现是在kuscia层实现的如果你需要尝试添加新的数据源配置可以优先看一下DataMesh的介绍。
https://www.secretflow.org.cn/zh-CN/docs/kuscia/v0.12.0b0/reference/apis/datamesh/datacrud_cn

SecretPad部分主要做了数据源配置的输入校验等功能，你可以服用或参照mysql中的配置创建自己的实现

[Trae-Young]

数据源类型添加，我看应该是有kuscia api和datamesh api。
kuscia api我是围绕proto/api/v1alpha1/kusciaapi/domaindatasource.proto和pkg/kusciaapi/service/domaindata_source.go两个文件进行改造的。
datamesh api则是围绕proto/api/v1alpha1/datamesh/domaindatasource.proto和pkg/datamesh/service/domaindatasource.go两个文件进行改造的。
但是我发现，kuscia层面的数据源，好像仅仅是一些连接信息，并没有涉及到对这些数据源信息DataSourceInfo进行一些jdbc等建立连接的操作。和数据源进行真正交互的是kuscia层面？还是secretpad?
我觉得单纯对kuscia添加数据源信息改造肯定是无法真正实现数据源类型的添加。

[wangzul]

数据源进行真正交互的是kuscia，具体可以参照下方链接来了解。
DataMesh的实现
https://github.com/secretflow/kuscia/tree/main/pkg/datamesh
DataProxy
https://github.com/secretflow/dataproxy

[Trae-Young]

好的，我这两天一直在看Kuscia的DataMesh和Kusciaapi中的domaindatasource相关，也看了DataProxy工具，但是这些看起来，添加一个新的类型数据源，比如oracle，我只需要对kuscia的proto文件和对应的pkg/kusciaapi/service/domaindata_source.go文件和datamesh对应的go文件进行修改，就可以实现在kuscia上添加数据源类型，在kuscia中还有其他地方要改造嘛？
关于dataproxy，这个项目的编译、打包镜像都可以实现，但是其应用方法并没有一个官方文档去说明，无法清楚地知道怎么使用这个dataproxy工具。

[wangzul]

对于 Kuscia 来说，需要在创建数据源的API 地方放开校验，在 DataProxy 或者 datamesh 中实现扩展数据源的 IO 操作。

dataproxy提供了 odps 的配置，可以根据需要，参照新增数据源实现

对于datamesh来说，需要修改的内容，kusciaapi、datameshapi相关的proto以外，就是参照mysql、oss等目前已有的数据源，要修改对数据源字段的校验（比如mysql、oss、local等内置支持的类型报错）。按照DataMeshDataIOInterface的定义，完成对自定义数据源读写流程的编写。

[Trae-Young]

明白，如果我暂时不关注dataproxy扩展数据源的操作，而是更侧重于在Kuscia添加数据源。`
现在我参考MySQL已经完成对Oracle数据源的添加（通过修改proto以及pkg/datamesh/dataserver/io/builtin目录内的DataIOInterface）,并且测试能够完成/api/v1/domaindatasource/create对oracle数据源的添加。
但是我又关注到自定义数据源在读写方面应该是有部分涉及到了数据类型转换，因为还没有对oracle数据源进行实际的读写操作测试，所以无法判断下面这部分代码的作用是否影响数据源的读写：
pkg/common/convert.go
func Convert2ArrowColumnType(colType string) arrow.DataType {
switch strings.ToLower(colType) {
case "int8":
return arrow.PrimitiveTypes.Int8
case "int16":
return arrow.PrimitiveTypes.Int16
case "int32":
return arrow.PrimitiveTypes.Int32
case "int64", "int":
return arrow.PrimitiveTypes.Int64
case "uint8":
return arrow.PrimitiveTypes.Uint8
case "uint16":
return arrow.PrimitiveTypes.Uint16
case "uint32":
return arrow.PrimitiveTypes.Uint32
case "uint64", "uint":
return arrow.PrimitiveTypes.Uint64
case "float32":
return arrow.PrimitiveTypes.Float32
case "float64", "float":
return arrow.PrimitiveTypes.Float64
case "date32":
return arrow.PrimitiveTypes.Date32
case "date64":
return arrow.PrimitiveTypes.Date64
case "bool":
return arrow.FixedWidthTypes.Boolean
// STRING UTF8 variable-length string as List
case "string", "str":
return arrow.BinaryTypes.String
// Variable-length bytes (no guarantee of UTF8-ness)
case "binary":
return arrow.BinaryTypes.Binary
}
return nil
}

func Convert2DataMeshColumnType(colType arrow.DataType) string {
switch colType {
case arrow.PrimitiveTypes.Int8:
return "int8"
case arrow.PrimitiveTypes.Int16:
return "int16"
case arrow.PrimitiveTypes.Int32:
return "int32"
case arrow.PrimitiveTypes.Int64:
return "int64"
case arrow.PrimitiveTypes.Uint8:
return "uint8"
case arrow.PrimitiveTypes.Uint16:
return "uint16"
case arrow.PrimitiveTypes.Uint32:
return "uint32"
case arrow.PrimitiveTypes.Uint64:
return "uint64"
case arrow.PrimitiveTypes.Float32:
return "float32"
case arrow.PrimitiveTypes.Float64:
return "float64"
case arrow.PrimitiveTypes.Date32:
return "date32"
case arrow.PrimitiveTypes.Date64:
return "date64"
case arrow.FixedWidthTypes.Boolean:
return "bool"
// STRING UTF8 variable-length string as List
case arrow.BinaryTypes.String:
return "string"
// Variable-length bytes (no guarantee of UTF8-ness)
case arrow.BinaryTypes.Binary:
return "binary"
}
return ""
}

以及下列代码pkg/datamesh/metaserver/service/domaindata.go

func CheckColType(cols []*pbv1alpha1.DataColumn, dsType string) error {
for _, col := range cols {
col.Type = strings.ToLower(col.Type)
if dsType == common.DomainDataSourceTypeMysql {
switch col.Type {
case "int", "int8", "int16", "int32", "int64":
case "uint", "uint8", "uint16", "uint32", "uint64":
case "float", "float32", "float64":
case "bool":
case "string", "str":
return nil
default:
err := fmt.Errorf("Col[%s].Type=%s is invalid for mysql", col.Name, col.Type)
nlog.Error(err)
return err
}
return nil
}
// 判断 Oracle 数据源类型
if dsType == common.DomainDataSourceTypeOracle {
switch col.Type {
case "number", "integer", "float", "decimal":
case "varchar2", "char", "clob":
case "boolean": // 使用 NUMBER(1) 来表示布尔值
case "date", "timestamp", "timestamp with time zone":
case "interval", "interval year to month", "interval day to second":
case "blob", "raw": // 二进制数据类型
case "rowid", "urowid": // 行标识符
return nil
default:
err := fmt.Errorf("Col[%s].Type=%s is invalid for oracle", col.Name, col.Type)
nlog.Error(err)
return err
}
return nil
}

	// Hive 数据库类型验证
	//if dsType == common.DomainDataSourceTypeHive {
	//	switch col.Type {
	//	case "int", "bigint", "tinyint", "smallint":
	//	case "float", "double":
	//	case "boolean":
	//	case "string", "varchar", "char":
	//	case "timestamp", "date":
	//	case "array", "map", "struct", "uniontype":
	//	case "decimal":
	//	case "binary", "varbinary":
	//		return nil
	//	default:
	//		err := fmt.Errorf("Col[%s].Type=%s is invalid for hive", col.Name, col.Type)
	//		nlog.Error(err)
	//		return err
	//	}
	//	return nil
	//}

	arrowType := common.Convert2ArrowColumnType(col.Type)
	if arrowType == nil {
		err := fmt.Errorf("Col[%s].Type=%s is invalid for DataProxy", col.Name, col.Type)
		nlog.Error(err)
		return err
	}
}
return nil

}

[wangzul]

kuscia目前还未支持Oracle数据源，并没有做过系统性验证，不好对您的代码做出更好的说明。
你可以尝试运行，如果有异常我们会协助你解决这个问题。

[Trae-Young]

在secretpad all-in-one部署模式下，如果想进行kuscia数据源系统性验证的话，是不是secretpad也需要做Oracle数据源类型的添加改造？然后再替换镜像？
两者除了同步proto文件之外，secretpad还需要参考kuscia关于数据源改造部分的代码吗？

[github-actions]

Stale issue message. Please comment to remove stale tag. Otherwise this issue will be closed soon.

[oceanqdu]

@Trae-Young 您好，请问您成功实现kusica添加数据源了吗？我也有类似的需求想请教一下

[Trae-Young]

@Trae-Young 您好，请问您成功实现kusica添加数据源了吗？我也有类似的需求想请教一下

我只是自己尝试了增加数据源，通过api接口的调用可以实现对数据源类型的创建，但是落到实际的应用场景进行数据源的数据交互，还未进行系统性测试。

[Candicepan]

也可以参考这个 PR：Kuscia DataMesh 支持 PostgreSQL 数据源 #672

[YanZhuangz]

@Trae-Young 您好，请问您成功实现kusica添加数据源了吗？我也有类似的需求想请教一下

@oceanqdu 目前初步提供了数据源扩展的指导文档，不太完善，可以帮忙提一下建议，有不明白的可以另外提一个 issue 沟通，对文档进行进一步补充优化。如何扩展其他数据源