diff --git a/api/all.yaml b/api/all.yaml index 9e4f95268d..547aed5989 100644 --- a/api/all.yaml +++ b/api/all.yaml @@ -1678,6 +1678,7 @@ components: - EXTERNAL - STREAMING_TABLE # Not yet fully implemented - MATERIALIZED_VIEW # Not yet fully implemented + - METRIC_VIEW DataSourceFormat: description: Data source format type: string @@ -1738,6 +1739,11 @@ components: table_id: description: Unique identifier for the table. type: string + view_definition: + description: SQL query that defines the metric view (for METRIC_VIEW table type). + type: string + view_dependencies: + "$ref": "#/components/schemas/DependencyList" CreateTable: type: object properties: @@ -1767,14 +1773,17 @@ components: type: string properties: $ref: '#/components/schemas/SecurablePropertiesMap' + view_definition: + description: SQL query that defines the metric view (for METRIC_VIEW table type). + type: string + view_dependencies: + "$ref": "#/components/schemas/DependencyList" required: - name - catalog_name - schema_name - table_type - - data_source_format - columns - - storage_location StagingTableInfo: type: object properties: diff --git a/build.sbt b/build.sbt index 9cafdf9ecf..efddadbf25 100644 --- a/build.sbt +++ b/build.sbt @@ -594,6 +594,10 @@ lazy val spark = (project in file("connectors/spark")) lombokPath ) }, + Compile / unmanagedJars ++= { + val sparkAssemblyDir = sys.props.get("sparkAssemblyDir").map(file).filter(_.exists) + sparkAssemblyDir.toSeq.flatMap(d => (d ** "*.jar").get.classpath) + }, libraryDependencies ++= Seq( "org.apache.spark" %% "spark-sql" % sparkVersion % Provided, "com.fasterxml.jackson.core" % "jackson-databind" % "2.15.0", @@ -619,8 +623,9 @@ lazy val spark = (project in file("connectors/spark")) "org.apache.hadoop" % "hadoop-aws" % hadoopVersion % Test, "org.projectlombok" % "lombok" % "1.18.32" % Test, "com.google.cloud.bigdataoss" % "gcs-connector" % "3.0.2" % Test classifier "shaded", + ) ++ (if (sparkMajorMinorVersion == "4.0") Seq( "io.delta" %% s"delta-spark_$sparkMajorMinorVersion" % deltaVersion % Test, - ), + ) else Nil), dependencyOverrides ++= Seq( "com.fasterxml.jackson.core" % "jackson-databind" % "2.15.0", "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.15.0", @@ -676,7 +681,9 @@ lazy val integrationTests = (project in file("integration-tests")) "org.assertj" % "assertj-core" % "3.26.3" % Test, "org.projectlombok" % "lombok" % "1.18.32" % Provided, "org.apache.spark" %% "spark-sql" % sparkVersion % Test, + ) ++ (if (sparkMajorMinorVersion == "4.0") Seq( "io.delta" %% s"delta-spark_$sparkMajorMinorVersion" % deltaVersion % Test, + ) else Nil) ++ Seq( "org.apache.hadoop" % "hadoop-aws" % hadoopVersion % Test, "org.apache.hadoop" % "hadoop-azure" % hadoopVersion % Test, "com.google.cloud.bigdataoss" % "gcs-connector" % "3.0.2" % Test classifier "shaded", diff --git a/connectors/spark/src/main/scala/io/unitycatalog/spark/UCSingleCatalog.scala b/connectors/spark/src/main/scala/io/unitycatalog/spark/UCSingleCatalog.scala index 1d3daca7f3..19b633a951 100644 --- a/connectors/spark/src/main/scala/io/unitycatalog/spark/UCSingleCatalog.scala +++ b/connectors/spark/src/main/scala/io/unitycatalog/spark/UCSingleCatalog.scala @@ -99,6 +99,13 @@ class UCSingleCatalog delegate.tableExists(ident) } + override def createTable( + ident: Identifier, + tableInfo: org.apache.spark.sql.connector.catalog.TableInfo): Table = { + UCSingleCatalog.checkUnsupportedNestedNamespace(ident.namespace()) + delegate.createTable(ident, tableInfo) + } + override def createTable( ident: Identifier, columns: Array[Column], @@ -582,7 +589,10 @@ private class UCProxy( case e: ApiException if e.getCode == 404 => throw new NoSuchTableException(ident) } + val identifier = TableIdentifier(t.getName, Some(t.getSchemaName), Some(t.getCatalogName)) + val hasStorage = t.getStorageLocation != null + val partitionCols = scala.collection.mutable.ArrayBuffer.empty[(String, Int)] val fields = t.getColumns.asScala.map { col => Option(col.getPartitionIndex).foreach { index => @@ -591,69 +601,85 @@ private class UCProxy( StructField(col.getName, DataType.fromDDL(col.getTypeText), col.getNullable) .withComment(col.getComment) }.toArray - val locationUri = CatalogUtils.stringToURI(t.getStorageLocation) - val tableId = t.getTableId - var tableOp = TableOperation.READ_WRITE - val temporaryCredentials = { - try { - temporaryCredentialsApi - .generateTemporaryTableCredentials( - // TODO: at this time, we don't know if the table will be read or written. For now we always - // request READ_WRITE credentials as the server doesn't distinguish between READ and - // READ_WRITE credentials as of today. When loading a table, Spark should tell if it's - // for read or write, we can request the proper credential after fixing Spark. - new GenerateTemporaryTableCredential().tableId(tableId).operation(tableOp) - ) - } catch { - case e: ApiException => - logWarning(s"READ_WRITE credential generation failed for table $identifier: ${e.getMessage}") - try { - tableOp = TableOperation.READ - temporaryCredentialsApi - .generateTemporaryTableCredentials( - new GenerateTemporaryTableCredential().tableId(tableId).operation(tableOp) - ) - } catch { - case e: ApiException => - logWarning(s"READ credential generation failed for table $identifier: ${e.getMessage}") - if (serverSidePlanningEnabled) null else throw e - } + + // Credential vending -- only for tables with storage + val (storage, extraProps) = if (hasStorage) { + val locationUri = CatalogUtils.stringToURI(t.getStorageLocation) + val tableId = t.getTableId + var tableOp = TableOperation.READ_WRITE + val temporaryCredentials = { + try { + temporaryCredentialsApi + .generateTemporaryTableCredentials( + // TODO: at this time, we don't know if the table will be read or written. For now we always + // request READ_WRITE credentials as the server doesn't distinguish between READ and + // READ_WRITE credentials as of today. When loading a table, Spark should tell if it's + // for read or write, we can request the proper credential after fixing Spark. + new GenerateTemporaryTableCredential().tableId(tableId).operation(tableOp) + ) + } catch { + case e: ApiException => + logWarning(s"READ_WRITE credential generation failed for table $identifier: ${e.getMessage}") + try { + tableOp = TableOperation.READ + temporaryCredentialsApi + .generateTemporaryTableCredentials( + new GenerateTemporaryTableCredential().tableId(tableId).operation(tableOp) + ) + } catch { + case e: ApiException => + logWarning(s"READ credential generation failed for table $identifier: ${e.getMessage}") + if (serverSidePlanningEnabled) null else throw e + } + } } - } - if (serverSidePlanningEnabled && temporaryCredentials == null) { - enableServerSidePlanningConfig(identifier) - } + if (serverSidePlanningEnabled && temporaryCredentials == null) { + enableServerSidePlanningConfig(identifier) + } - val extraSerdeProps = if (temporaryCredentials == null) { - Map.empty[String, String].asJava - } else { - CredPropsUtil.createTableCredProps( - renewCredEnabled, - credScopedFsEnabled, - UCSingleCatalog.sessionHadoopFsImplProps(), - locationUri.getScheme, - uri.toString, - tokenProvider, - tableId, - tableOp, - temporaryCredentials, + val extraSerdeProps = if (temporaryCredentials == null) { + Map.empty[String, String].asJava + } else { + CredPropsUtil.createTableCredProps( + renewCredEnabled, + credScopedFsEnabled, + UCSingleCatalog.sessionHadoopFsImplProps(), + locationUri.getScheme, + uri.toString, + tokenProvider, + tableId, + tableOp, + temporaryCredentials, + ) + } + + val storageFormat = CatalogStorageFormat.empty.copy( + locationUri = Some(locationUri), + properties = t.getProperties.asScala.toMap ++ extraSerdeProps ) + (storageFormat, Map.empty[String, String]) + } else { + (CatalogStorageFormat.empty, + Option(t.getProperties).map(_.asScala.toMap).getOrElse(Map.empty)) + } + + // Table type mapping + val catalogTableType = t.getTableType match { + case TableType.MANAGED => CatalogTableType.MANAGED + case TableType.METRIC_VIEW => CatalogTableType.METRIC_VIEW + case _ => CatalogTableType.EXTERNAL } val sparkTable = CatalogTable( identifier, - tableType = if (t.getTableType == TableType.MANAGED) { - CatalogTableType.MANAGED - } else { - CatalogTableType.EXTERNAL - }, - storage = CatalogStorageFormat.empty.copy( - locationUri = Some(locationUri), - properties = t.getProperties.asScala.toMap ++ extraSerdeProps - ), + tableType = catalogTableType, + storage = storage, schema = StructType(fields), - provider = Some(t.getDataSourceFormat.getValue.toLowerCase()), + provider = Option(t.getDataSourceFormat).map(_.getValue.toLowerCase()), + viewText = Option(t.getViewDefinition), + viewOriginalText = Option(t.getViewDefinition), + properties = extraProps, createTime = t.getCreatedAt, tracksPartitionsInCatalog = false, partitionColumnNames = partitionCols.sortBy(_._2).map(_._1).toSeq @@ -667,33 +693,102 @@ private class UCProxy( .asInstanceOf[Table] } - override def createTable(ident: Identifier, schema: StructType, partitions: Array[Transform], properties: util.Map[String, String]): Table = { + override def createTable( + ident: Identifier, + tableInfo: org.apache.spark.sql.connector.catalog.TableInfo): Table = { UCSingleCatalog.checkUnsupportedNestedNamespace(ident.namespace()) - UCSingleCatalog.requireProviderSpecified("CREATE TABLE", properties) + val properties = tableInfo.properties() + val ct = new CreateTable() + ct.setName(ident.name()) + ct.setSchemaName(ident.namespace().head) + ct.setCatalogName(this.name) + Option(properties.get(TableCatalog.PROP_COMMENT)).foreach(ct.setComment(_)) + val serverProps = + properties.view.filterKeys(!UCTableProperties.V2_TABLE_PROPERTIES.contains(_)).toMap + ct.setProperties(serverProps) - val createTable = new CreateTable() - createTable.setName(ident.name()) - createTable.setSchemaName(ident.namespace().head) - createTable.setCatalogName(this.name) + // Table type: use explicit tableType if provided, otherwise infer from properties + Option(tableInfo.tableType()) match { + case Some(tt) => ct.setTableType(TableType.fromValue(tt)) + case None => + val isManagedLocation = Option(properties.get(TableCatalog.PROP_IS_MANAGED_LOCATION)) + .exists(_.equalsIgnoreCase("true")) + if (isManagedLocation) { + ct.setTableType(TableType.MANAGED) + } else { + ct.setTableType(TableType.EXTERNAL) + } + } - val hasExternalClause = properties.containsKey(TableCatalog.PROP_EXTERNAL) - val storageLocation = properties.get(TableCatalog.PROP_LOCATION) - assert(storageLocation != null, "location should either be user specified or system generated.") - val isManagedLocation = Option(properties.get(TableCatalog.PROP_IS_MANAGED_LOCATION)) - .exists(_.equalsIgnoreCase("true")) - val format = properties.get("provider") - if (isManagedLocation) { - assert(!hasExternalClause, "location is only generated for managed tables.") - if (!format.equalsIgnoreCase(DataSourceFormat.DELTA.name)) { - throw new ApiException("Unity Catalog does not support non-Delta managed table.") + // Storage location (null for metric views and other view types) + Option(properties.get(TableCatalog.PROP_LOCATION)).foreach(ct.setStorageLocation(_)) + + // Data source format + Option(properties.get("provider")).foreach { format => + ct.setDataSourceFormat(convertDatasourceFormat(format)) + } + + // Columns + val partitionColNames = extractPartitionColumnNames(tableInfo.partitions()) + val columns: Seq[ColumnInfo] = tableInfo.columns().toSeq.zipWithIndex.map { case (col, i) => + val column = new ColumnInfo() + column.setName(col.name()) + column.setNullable(col.nullable()) + column.setTypeText(col.dataType().catalogString) + column.setTypeName(convertDataTypeToTypeName(col.dataType())) + column.setTypeJson(col.dataType().json) + column.setPosition(i) + Option(col.comment()).foreach(column.setComment(_)) + val partitionIdx = partitionColNames.indexWhere(_.equalsIgnoreCase(col.name())) + if (partitionIdx >= 0) column.setPartitionIndex(partitionIdx) + column + } + ct.setColumns(columns.asJava) + + // View definition and dependencies (null for non-view tables) + Option(tableInfo.viewDefinition()).foreach(ct.setViewDefinition(_)) + Option(tableInfo.viewDependencies()).foreach { sparkDepList => + val ucDepList = new io.unitycatalog.client.model.DependencyList() + val ucDeps = sparkDepList.dependencies().map { dep => + val ucDep = new io.unitycatalog.client.model.Dependency() + dep match { + case td: org.apache.spark.sql.connector.catalog.TableDependency => + ucDep.setTable( + new io.unitycatalog.client.model.TableDependency() + .tableFullName(td.tableFullName())) + case fd: org.apache.spark.sql.connector.catalog.FunctionDependency => + ucDep.setFunction( + new io.unitycatalog.client.model.FunctionDependency() + .functionFullName(fd.functionFullName())) + case _ => + } + ucDep } - createTable.setTableType(TableType.MANAGED) - } else { - createTable.setTableType(TableType.EXTERNAL) + ucDepList.setDependencies(java.util.Arrays.asList(ucDeps: _*)) + ct.setViewDependencies(ucDepList) + } + + tablesApi.createTable(ct) + loadTable(ident) + } + + override def createTable(ident: Identifier, schema: StructType, partitions: Array[Transform], properties: util.Map[String, String]): Table = { + UCSingleCatalog.checkUnsupportedNestedNamespace(ident.namespace()) + UCSingleCatalog.requireProviderSpecified("CREATE TABLE", properties) + val columns: Array[Column] = schema.fields.map { f => + Column.create(f.name, f.dataType, f.nullable, f.getComment().orNull, null) } - createTable.setStorageLocation(storageLocation) + val tableInfo = new org.apache.spark.sql.connector.catalog.TableInfo.Builder() + .withColumns(columns) + .withPartitions(partitions) + .withProperties(properties) + .build() + createTable(ident, tableInfo) + } - val partitionColNames: Seq[String] = partitions.flatMap { t => + private def extractPartitionColumnNames(partitions: Array[Transform]): Seq[String] = { + if (partitions == null) return Seq.empty + partitions.flatMap { t => t.name() match { case "identity" => val fieldNames = t.references().flatMap(_.fieldNames()) @@ -706,31 +801,6 @@ private class UCProxy( throw new ApiException(s"Unsupported partition transform: $other") } }.toSeq - val columns: Seq[ColumnInfo] = schema.fields.toSeq.zipWithIndex.map { case (field, i) => - val column = new ColumnInfo() - column.setName(field.name) - if (field.getComment().isDefined) { - column.setComment(field.getComment.get) - } - column.setNullable(field.nullable) - column.setTypeText(field.dataType.catalogString) - column.setTypeName(convertDataTypeToTypeName(field.dataType)) - column.setTypeJson(field.dataType.json) - column.setPosition(i) - val partitionIdx = partitionColNames.indexWhere(_.equalsIgnoreCase(field.name)) - if (partitionIdx >= 0) column.setPartitionIndex(partitionIdx) - column - } - val comment = Option(properties.get(TableCatalog.PROP_COMMENT)) - comment.foreach(createTable.setComment(_)) - createTable.setColumns(columns) - createTable.setDataSourceFormat(convertDatasourceFormat(format)) - // Do not send the V2 table properties as they are made part of the `createTable` already. - val propertiesToServer = - properties.view.filterKeys(!UCTableProperties.V2_TABLE_PROPERTIES.contains(_)).toMap - createTable.setProperties(propertiesToServer) - tablesApi.createTable(createTable) - loadTable(ident) } private def convertDatasourceFormat(format: String): DataSourceFormat = { diff --git a/server/src/main/java/io/unitycatalog/server/persist/DependencyRepository.java b/server/src/main/java/io/unitycatalog/server/persist/DependencyRepository.java new file mode 100644 index 0000000000..0cb33b8b93 --- /dev/null +++ b/server/src/main/java/io/unitycatalog/server/persist/DependencyRepository.java @@ -0,0 +1,56 @@ +package io.unitycatalog.server.persist; + +import io.unitycatalog.server.persist.dao.DependencyDAO; +import java.util.List; +import java.util.UUID; +import org.hibernate.Session; +import org.hibernate.query.Query; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Repository for managing view/metric-view dependency records in uc_dependencies. Methods accept a + * Session so they can participate in the caller's transaction. + */ +public class DependencyRepository { + private static final Logger LOGGER = LoggerFactory.getLogger(DependencyRepository.class); + + public void createDependencies( + Session session, UUID dependentId, String dependentType, List dependencies) { + for (DependencyDAO dep : dependencies) { + dep.setDependentId(dependentId); + dep.setDependentType(dependentType); + session.persist(dep); + } + LOGGER.debug( + "Created {} dependencies for {}:{}", dependencies.size(), dependentType, dependentId); + } + + public List getDependencies( + Session session, UUID dependentId, String dependentType) { + String hql = + "FROM DependencyDAO d WHERE d.dependentId = :dependentId" + + " AND d.dependentType = :dependentType"; + Query query = session.createQuery(hql, DependencyDAO.class); + query.setParameter("dependentId", dependentId); + query.setParameter("dependentType", dependentType); + return query.list(); + } + + public void deleteDependencies(Session session, UUID dependentId, String dependentType) { + String hql = + "DELETE FROM DependencyDAO d WHERE d.dependentId = :dependentId" + + " AND d.dependentType = :dependentType"; + Query query = session.createQuery(hql); + query.setParameter("dependentId", dependentId); + query.setParameter("dependentType", dependentType); + int deleted = query.executeUpdate(); + LOGGER.debug("Deleted {} dependencies for {}:{}", deleted, dependentType, dependentId); + } + + public void updateDependencies( + Session session, UUID dependentId, String dependentType, List dependencies) { + deleteDependencies(session, dependentId, dependentType); + createDependencies(session, dependentId, dependentType, dependencies); + } +} diff --git a/server/src/main/java/io/unitycatalog/server/persist/Repositories.java b/server/src/main/java/io/unitycatalog/server/persist/Repositories.java index 7222450e5b..ac4e6f4964 100644 --- a/server/src/main/java/io/unitycatalog/server/persist/Repositories.java +++ b/server/src/main/java/io/unitycatalog/server/persist/Repositories.java @@ -29,6 +29,7 @@ public class Repositories { private final CredentialRepository credentialRepository; private final ExternalLocationRepository externalLocationRepository; private final DeltaCommitRepository deltaCommitRepository; + private final DependencyRepository dependencyRepository; private final KeyMapper keyMapper; @@ -50,6 +51,7 @@ public Repositories(SessionFactory sessionFactory, ServerProperties serverProper this.credentialRepository = new CredentialRepository(this, sessionFactory, serverProperties); this.externalLocationRepository = new ExternalLocationRepository(this, sessionFactory); this.deltaCommitRepository = new DeltaCommitRepository(sessionFactory, serverProperties); + this.dependencyRepository = new DependencyRepository(); // KeyMapper uses all the repositories above. this.keyMapper = new KeyMapper(this); diff --git a/server/src/main/java/io/unitycatalog/server/persist/TableRepository.java b/server/src/main/java/io/unitycatalog/server/persist/TableRepository.java index cf30f20efb..93006a79cc 100644 --- a/server/src/main/java/io/unitycatalog/server/persist/TableRepository.java +++ b/server/src/main/java/io/unitycatalog/server/persist/TableRepository.java @@ -5,9 +5,11 @@ import io.unitycatalog.server.model.ColumnInfo; import io.unitycatalog.server.model.CreateTable; import io.unitycatalog.server.model.DataSourceFormat; +import io.unitycatalog.server.model.DependencyList; import io.unitycatalog.server.model.ListTablesResponse; import io.unitycatalog.server.model.TableInfo; import io.unitycatalog.server.model.TableType; +import io.unitycatalog.server.persist.dao.DependencyDAO; import io.unitycatalog.server.persist.dao.PropertyDAO; import io.unitycatalog.server.persist.dao.SchemaInfoDAO; import io.unitycatalog.server.persist.dao.StagingTableDAO; @@ -151,6 +153,7 @@ public TableInfo getTable(String fullName) { TableInfo tableInfo = tableInfoDAO.toTableInfo(true, catalogName, schemaName); RepositoryUtils.attachProperties( tableInfo, tableInfo.getTableId(), Constants.TABLE, session); + attachDependencies(tableInfo, tableInfoDAO, session); return tableInfo; }, "Failed to get table", @@ -190,7 +193,6 @@ public TableInfo createTable(CreateTable createTable) { repositories .getSchemaRepository() .getSchemaIdOrThrow(session, catalogName, schemaName); - NormalizedURL storageLocation = NormalizedURL.from(createTable.getStorageLocation()); // Check if table already exists TableInfoDAO existingTable = @@ -200,13 +202,14 @@ public TableInfo createTable(CreateTable createTable) { ErrorCode.TABLE_ALREADY_EXISTS, "Table already exists: " + fullName); } TableType tableType = Objects.requireNonNull(createTable.getTableType()); - // The table ID will either be a new random one or the id of staging table, depending - // on the type of table to create. String tableID; + NormalizedURL storageLocation; if (tableType == TableType.EXTERNAL) { + storageLocation = NormalizedURL.from(createTable.getStorageLocation()); ExternalLocationUtils.validateNotOverlapWithManagedStorage(session, storageLocation); tableID = UUID.randomUUID().toString(); } else if (tableType == TableType.MANAGED) { + storageLocation = NormalizedURL.from(createTable.getStorageLocation()); serverProperties.checkManagedTableEnabled(); if (createTable.getDataSourceFormat() != DataSourceFormat.DELTA) { throw new BaseException( @@ -219,6 +222,31 @@ public TableInfo createTable(CreateTable createTable) { .getStagingTableRepository() .commitStagingTable(session, callerId, storageLocation); tableID = stagingTableDAO.getId().toString(); + } else if (tableType == TableType.METRIC_VIEW) { + if (createTable.getViewDefinition() == null + || createTable.getViewDefinition().isEmpty()) { + throw new BaseException( + ErrorCode.INVALID_ARGUMENT, "view_definition is required for metric view"); + } + DependencyList viewDeps = createTable.getViewDependencies(); + if (viewDeps == null + || viewDeps.getDependencies() == null + || viewDeps.getDependencies().isEmpty()) { + throw new BaseException( + ErrorCode.INVALID_ARGUMENT, "view_dependencies is required for metric view"); + } + storageLocation = null; + tableID = UUID.randomUUID().toString(); + if (viewDeps.getDependencies() != null) { + UUID tableUUID = UUID.fromString(tableID); + List depDAOs = + viewDeps.getDependencies().stream() + .map(dep -> DependencyDAO.from(dep, tableUUID, "TABLE")) + .collect(Collectors.toList()); + repositories + .getDependencyRepository() + .createDependencies(session, tableUUID, "TABLE", depDAOs); + } } else if (tableType == TableType.STREAMING_TABLE) { throw new BaseException( ErrorCode.INVALID_ARGUMENT, "STREAMING TABLE creation is not supported yet."); @@ -245,7 +273,8 @@ public TableInfo createTable(CreateTable createTable) { .createdBy(callerId) .updatedAt(createTime) .updatedBy(callerId) - .storageLocation(storageLocation.toString()) + .storageLocation(storageLocation != null ? storageLocation.toString() : null) + .viewDefinition(createTable.getViewDefinition()) .tableId(tableID); TableInfoDAO tableInfoDAO = TableInfoDAO.from(tableInfo, schemaId); @@ -342,6 +371,7 @@ public ListTablesResponse listTables( RepositoryUtils.attachProperties( tableInfo, tableInfo.getTableId(), Constants.TABLE, session); } + attachDependencies(tableInfo, tableInfoDAO, session); result.add(tableInfo); } return new ListTablesResponse().tables(result).nextPageToken(nextPageToken); @@ -384,8 +414,26 @@ public void deleteTable(Session session, UUID schemaId, String tableName) { .getDeltaCommitRepository() .permanentlyDeleteTableCommits(session, tableInfoDAO.getId()); } + if ("METRIC_VIEW".equals(tableInfoDAO.getType())) { + repositories + .getDependencyRepository() + .deleteDependencies(session, tableInfoDAO.getId(), "TABLE"); + } PropertyRepository.findProperties(session, tableInfoDAO.getId(), Constants.TABLE) .forEach(session::remove); session.remove(tableInfoDAO); } + + private void attachDependencies(TableInfo tableInfo, TableInfoDAO tableInfoDAO, Session session) { + if ("METRIC_VIEW".equals(tableInfoDAO.getType())) { + List deps = + repositories + .getDependencyRepository() + .getDependencies(session, tableInfoDAO.getId(), "TABLE"); + if (!deps.isEmpty()) { + tableInfo.setViewDependencies( + new DependencyList().dependencies(DependencyDAO.toDependencyList(deps))); + } + } + } } diff --git a/server/src/main/java/io/unitycatalog/server/persist/dao/DependencyDAO.java b/server/src/main/java/io/unitycatalog/server/persist/dao/DependencyDAO.java new file mode 100644 index 0000000000..b1872bad05 --- /dev/null +++ b/server/src/main/java/io/unitycatalog/server/persist/dao/DependencyDAO.java @@ -0,0 +1,118 @@ +package io.unitycatalog.server.persist.dao; + +import io.unitycatalog.server.model.Dependency; +import io.unitycatalog.server.model.TableDependency; +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.Index; +import jakarta.persistence.Table; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.ToString; +import org.hibernate.annotations.UuidGenerator; + +@Entity +@Table( + name = "uc_dependencies", + indexes = { + @Index(name = "idx_dependent", columnList = "dependent_type,dependent_id"), + }) +@Getter +@Setter +@NoArgsConstructor +@AllArgsConstructor +@EqualsAndHashCode +@ToString +@Builder +public class DependencyDAO { + @Id + @UuidGenerator + @Column(name = "id", updatable = false, nullable = false) + private UUID id; + + @Column(name = "dependent_type", nullable = false) + private String dependentType; + + @Column(name = "dependent_id", nullable = false) + private UUID dependentId; + + @Column(name = "dependency_type", nullable = false) + private String dependencyType; + + @Column(name = "dependency_catalog") + private String dependencyCatalog; + + @Column(name = "dependency_schema") + private String dependencySchema; + + @Column(name = "dependency_name") + private String dependencyName; + + @Column(name = "dependency_target_id") + private UUID dependencyTargetId; + + /** Converts a Dependency API model to a DependencyDAO for a given dependent. */ + public static DependencyDAO from(Dependency dependency, UUID dependentId, String dependentType) { + DependencyDAOBuilder builder = + DependencyDAO.builder().dependentId(dependentId).dependentType(dependentType); + + if (dependency.getTable() != null) { + builder.dependencyType("TABLE"); + String fullName = dependency.getTable().getTableFullName(); + String[] parts = fullName.split("\\."); + if (parts.length == 3) { + builder.dependencyCatalog(parts[0]); + builder.dependencySchema(parts[1]); + builder.dependencyName(parts[2]); + } else { + builder.dependencyName(fullName); + } + } else if (dependency.getFunction() != null) { + builder.dependencyType("FUNCTION"); + String fullName = dependency.getFunction().getFunctionFullName(); + String[] parts = fullName.split("\\."); + if (parts.length == 3) { + builder.dependencyCatalog(parts[0]); + builder.dependencySchema(parts[1]); + builder.dependencyName(parts[2]); + } else { + builder.dependencyName(fullName); + } + } + + return builder.build(); + } + + /** Converts this DAO to a Dependency API model. */ + public Dependency toDependency() { + Dependency dependency = new Dependency(); + if ("TABLE".equals(dependencyType)) { + String fullName = dependencyCatalog + "." + dependencySchema + "." + dependencyName; + dependency.setTable(new TableDependency().tableFullName(fullName)); + } + return dependency; + } + + public String getDependencyFullName() { + if (dependencyCatalog != null && dependencySchema != null && dependencyName != null) { + return dependencyCatalog + "." + dependencySchema + "." + dependencyName; + } + return dependencyName; + } + + public static List toDependencyList(List daos) { + if (daos == null) { + return new ArrayList<>(); + } + return daos.stream().map(DependencyDAO::toDependency).collect(Collectors.toList()); + } +} diff --git a/server/src/main/java/io/unitycatalog/server/persist/dao/TableInfoDAO.java b/server/src/main/java/io/unitycatalog/server/persist/dao/TableInfoDAO.java index 7734f116e5..e630b26f7f 100644 --- a/server/src/main/java/io/unitycatalog/server/persist/dao/TableInfoDAO.java +++ b/server/src/main/java/io/unitycatalog/server/persist/dao/TableInfoDAO.java @@ -76,6 +76,9 @@ public class TableInfoDAO extends IdentifiableDAO { fetch = FetchType.LAZY) private List columns; + @Column(name = "view_definition", length = 65535) + private String viewDefinition; + @Column(name = "uniform_iceberg_metadata_location", length = 65535) private String uniformIcebergMetadataLocation; @@ -98,8 +101,12 @@ public static TableInfoDAO from(TableInfo tableInfo, UUID schemaId) { .updatedBy(tableInfo.getUpdatedBy()) .columnCount(tableInfo.getColumns() != null ? tableInfo.getColumns().size() : 0) .type(tableInfo.getTableType().toString()) - .dataSourceFormat(tableInfo.getDataSourceFormat().toString()) + .dataSourceFormat( + tableInfo.getDataSourceFormat() != null + ? tableInfo.getDataSourceFormat().toString() + : null) .url(tableInfo.getStorageLocation()) + .viewDefinition(tableInfo.getViewDefinition()) .columns(ColumnInfoDAO.fromList(tableInfo.getColumns())) .schemaId(schemaId) .build(); @@ -113,8 +120,10 @@ public TableInfo toTableInfo(boolean fetchColumns, String catalogName, String sc .catalogName(catalogName) .schemaName(schemaName) .tableType(TableType.valueOf(type)) - .dataSourceFormat(DataSourceFormat.valueOf(dataSourceFormat)) - .storageLocation(NormalizedURL.normalize(url)) + .dataSourceFormat( + dataSourceFormat != null ? DataSourceFormat.valueOf(dataSourceFormat) : null) + .storageLocation(url != null ? NormalizedURL.normalize(url) : null) + .viewDefinition(viewDefinition) .comment(comment) .owner(owner) .createdAt(createdAt != null ? createdAt.getTime() : null) diff --git a/server/src/main/java/io/unitycatalog/server/persist/utils/HibernateConfigurator.java b/server/src/main/java/io/unitycatalog/server/persist/utils/HibernateConfigurator.java index e3ee6672a3..486828bd5a 100644 --- a/server/src/main/java/io/unitycatalog/server/persist/utils/HibernateConfigurator.java +++ b/server/src/main/java/io/unitycatalog/server/persist/utils/HibernateConfigurator.java @@ -4,6 +4,7 @@ import io.unitycatalog.server.persist.dao.ColumnInfoDAO; import io.unitycatalog.server.persist.dao.CredentialDAO; import io.unitycatalog.server.persist.dao.DeltaCommitDAO; +import io.unitycatalog.server.persist.dao.DependencyDAO; import io.unitycatalog.server.persist.dao.ExternalLocationDAO; import io.unitycatalog.server.persist.dao.FunctionInfoDAO; import io.unitycatalog.server.persist.dao.FunctionParameterInfoDAO; @@ -71,6 +72,7 @@ private static SessionFactory createSessionFactory(Properties hibernatePropertie configuration.addAnnotatedClass(CredentialDAO.class); configuration.addAnnotatedClass(ExternalLocationDAO.class); configuration.addAnnotatedClass(DeltaCommitDAO.class); + configuration.addAnnotatedClass(DependencyDAO.class); ServiceRegistry serviceRegistry = new StandardServiceRegistryBuilder().applySettings(configuration.getProperties()).build(); diff --git a/server/src/test/java/io/unitycatalog/server/base/table/BaseMetricViewCRUDTest.java b/server/src/test/java/io/unitycatalog/server/base/table/BaseMetricViewCRUDTest.java new file mode 100644 index 0000000000..0d0a09cb6a --- /dev/null +++ b/server/src/test/java/io/unitycatalog/server/base/table/BaseMetricViewCRUDTest.java @@ -0,0 +1,172 @@ +package io.unitycatalog.server.base.table; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import io.unitycatalog.client.model.CreateTable; +import io.unitycatalog.client.model.Dependency; +import io.unitycatalog.client.model.DependencyList; +import io.unitycatalog.client.model.TableDependency; +import io.unitycatalog.client.model.TableInfo; +import io.unitycatalog.client.model.TableType; +import io.unitycatalog.server.utils.TestUtils; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.Test; + +public abstract class BaseMetricViewCRUDTest extends BaseTableCRUDTestEnv { + + protected static final String METRIC_VIEW_NAME = "uc_test_metric_view"; + protected static final String METRIC_VIEW_FULL_NAME = + TestUtils.CATALOG_NAME + "." + TestUtils.SCHEMA_NAME + "." + METRIC_VIEW_NAME; + protected static final String SOURCE_TABLE_FULL_NAME = + TestUtils.CATALOG_NAME + "." + TestUtils.SCHEMA_NAME + ".source_events"; + + protected static final String VIEW_DEFINITION_ASSET_SOURCE = + "version: \"0.1\"\n" + + "source: " + + SOURCE_TABLE_FULL_NAME + + "\n" + + "dimensions:\n" + + " - name: event_day\n" + + " expr: date_trunc('day', event_time)\n" + + "measures:\n" + + " - name: event_count\n" + + " expr: count(*)"; + + protected static final String VIEW_DEFINITION_SQL_SOURCE = + "version: \"0.1\"\n" + + "source: SELECT * FROM " + + SOURCE_TABLE_FULL_NAME + + " WHERE status = 'active'\n" + + "dimensions:\n" + + " - name: region\n" + + " expr: region\n" + + "measures:\n" + + " - name: total_amount\n" + + " expr: SUM(amount)"; + + protected static final Map PROPERTIES = + Map.of("team", "analytics", "refresh", "daily"); + + private static DependencyList makeDependencyList(String... tableFullNames) { + DependencyList depList = new DependencyList(); + depList.setDependencies( + java.util.Arrays.stream(tableFullNames) + .map(name -> new Dependency().table(new TableDependency().tableFullName(name))) + .collect(java.util.stream.Collectors.toList())); + return depList; + } + + @Test + public void testMetricViewCRUD() throws Exception { + assertThatThrownBy(() -> tableOperations.getTable(METRIC_VIEW_FULL_NAME)) + .isInstanceOf(Exception.class); + + // --- Create with asset source --- + CreateTable createRequest = + new CreateTable() + .name(METRIC_VIEW_NAME) + .catalogName(TestUtils.CATALOG_NAME) + .schemaName(TestUtils.SCHEMA_NAME) + .tableType(TableType.METRIC_VIEW) + .viewDefinition(VIEW_DEFINITION_ASSET_SOURCE) + .viewDependencies(makeDependencyList(SOURCE_TABLE_FULL_NAME)) + .comment("Daily event counts by day") + .properties(PROPERTIES); + + TableInfo created = tableOperations.createTable(createRequest); + assertThat(created.getName()).isEqualTo(METRIC_VIEW_NAME); + assertThat(created.getCatalogName()).isEqualTo(TestUtils.CATALOG_NAME); + assertThat(created.getSchemaName()).isEqualTo(TestUtils.SCHEMA_NAME); + assertThat(created.getTableType()).isEqualTo(TableType.METRIC_VIEW); + assertThat(created.getViewDefinition()).isEqualTo(VIEW_DEFINITION_ASSET_SOURCE); + assertThat(created.getTableId()).isNotNull(); + assertThat(created.getStorageLocation()) + .as("Metric views should have no storage location") + .isNull(); + + // --- Get and verify dependencies round-trip --- + TableInfo fetched = tableOperations.getTable(METRIC_VIEW_FULL_NAME); + assertThat(fetched.getName()).isEqualTo(METRIC_VIEW_NAME); + assertThat(fetched.getTableType()).isEqualTo(TableType.METRIC_VIEW); + assertThat(fetched.getViewDefinition()).isEqualTo(VIEW_DEFINITION_ASSET_SOURCE); + assertThat(fetched.getComment()).isEqualTo("Daily event counts by day"); + assertThat(fetched.getCreatedAt()).isNotNull(); + assertThat(fetched.getTableId()).isNotNull(); + assertThat(fetched.getViewDependencies()).isNotNull(); + assertThat(fetched.getViewDependencies().getDependencies()).hasSize(1); + assertThat(fetched.getViewDependencies().getDependencies().get(0).getTable().getTableFullName()) + .isEqualTo(SOURCE_TABLE_FULL_NAME); + + // Verify properties round-trip + assertThat(fetched.getProperties()).isNotNull(); + assertThat(fetched.getProperties().get("team")).isEqualTo("analytics"); + assertThat(fetched.getProperties().get("refresh")).isEqualTo("daily"); + + // --- List --- + List tables = + tableOperations.listTables(TestUtils.CATALOG_NAME, TestUtils.SCHEMA_NAME, Optional.empty()); + assertThat(tables) + .as("Metric view should appear in listTables") + .anyMatch( + t -> + METRIC_VIEW_NAME.equals(t.getName()) + && TableType.METRIC_VIEW.equals(t.getTableType())); + + // --- Delete --- + tableOperations.deleteTable(METRIC_VIEW_FULL_NAME); + assertThatThrownBy(() -> tableOperations.getTable(METRIC_VIEW_FULL_NAME)) + .isInstanceOf(Exception.class); + } + + @Test + public void testMetricViewWithSqlSource() throws Exception { + String otherTable = TestUtils.CATALOG_NAME + "." + TestUtils.SCHEMA_NAME + ".other_table"; + + CreateTable createRequest = + new CreateTable() + .name(METRIC_VIEW_NAME) + .catalogName(TestUtils.CATALOG_NAME) + .schemaName(TestUtils.SCHEMA_NAME) + .tableType(TableType.METRIC_VIEW) + .viewDefinition(VIEW_DEFINITION_SQL_SOURCE) + .viewDependencies(makeDependencyList(SOURCE_TABLE_FULL_NAME)) + .comment("Metric view with SQL source"); + + TableInfo created = tableOperations.createTable(createRequest); + assertThat(created.getTableType()).isEqualTo(TableType.METRIC_VIEW); + assertThat(created.getViewDefinition()).isEqualTo(VIEW_DEFINITION_SQL_SOURCE); + + TableInfo fetched = tableOperations.getTable(METRIC_VIEW_FULL_NAME); + assertThat(fetched.getViewDependencies()).isNotNull(); + assertThat(fetched.getViewDependencies().getDependencies()).hasSize(1); + + tableOperations.deleteTable(METRIC_VIEW_FULL_NAME); + } + + @Test + public void testCreateMetricViewWithoutDefinitionFails() throws Exception { + CreateTable badRequest = + new CreateTable() + .name(METRIC_VIEW_NAME) + .catalogName(TestUtils.CATALOG_NAME) + .schemaName(TestUtils.SCHEMA_NAME) + .tableType(TableType.METRIC_VIEW) + .viewDependencies(makeDependencyList(SOURCE_TABLE_FULL_NAME)); + assertThatThrownBy(() -> tableOperations.createTable(badRequest)).isInstanceOf(Exception.class); + } + + @Test + public void testCreateMetricViewWithoutDependenciesFails() throws Exception { + CreateTable badRequest = + new CreateTable() + .name(METRIC_VIEW_NAME) + .catalogName(TestUtils.CATALOG_NAME) + .schemaName(TestUtils.SCHEMA_NAME) + .tableType(TableType.METRIC_VIEW) + .viewDefinition(VIEW_DEFINITION_ASSET_SOURCE); + assertThatThrownBy(() -> tableOperations.createTable(badRequest)).isInstanceOf(Exception.class); + } +} diff --git a/server/src/test/java/io/unitycatalog/server/sdk/tables/SdkMetricViewCRUDTest.java b/server/src/test/java/io/unitycatalog/server/sdk/tables/SdkMetricViewCRUDTest.java new file mode 100644 index 0000000000..728611b9d9 --- /dev/null +++ b/server/src/test/java/io/unitycatalog/server/sdk/tables/SdkMetricViewCRUDTest.java @@ -0,0 +1,28 @@ +package io.unitycatalog.server.sdk.tables; + +import io.unitycatalog.server.base.ServerConfig; +import io.unitycatalog.server.base.catalog.CatalogOperations; +import io.unitycatalog.server.base.schema.SchemaOperations; +import io.unitycatalog.server.base.table.BaseMetricViewCRUDTest; +import io.unitycatalog.server.base.table.TableOperations; +import io.unitycatalog.server.sdk.catalog.SdkCatalogOperations; +import io.unitycatalog.server.sdk.schema.SdkSchemaOperations; +import io.unitycatalog.server.utils.TestUtils; + +public class SdkMetricViewCRUDTest extends BaseMetricViewCRUDTest { + + @Override + protected CatalogOperations createCatalogOperations(ServerConfig config) { + return new SdkCatalogOperations(TestUtils.createApiClient(config)); + } + + @Override + protected SchemaOperations createSchemaOperations(ServerConfig config) { + return new SdkSchemaOperations(TestUtils.createApiClient(config)); + } + + @Override + protected TableOperations createTableOperations(ServerConfig config) { + return new SdkTableOperations(TestUtils.createApiClient(config)); + } +}