aicam commented on code in PR #7762: URL: https://github.com/apache/texera/pull/7762#discussion_r3845498595
########## file-service/src/main/scala/org/apache/texera/service/type/LakeFSFileNode.scala: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.`type` + +import io.lakefs.clients.sdk.model.ObjectStats +import org.apache.texera.amber.core.storage.ResourceType + +import scala.collection.mutable + +// LakeFSFileNode represents a unique file in a versioned resource (dataset or model). +// Its full path is in the format of: +// /<resourceType>/ownerEmail/resourceName/versionName/fileRelativePath +// e.g. /dataset/[email protected]/twitterDataset/v1/california/irvine/tw1.csv +// /model/[email protected]/sentimentModel/v1/model.pt +class LakeFSFileNode( + val name: String, // direct name of this node + val nodeType: String, // "file" or "directory" + val parent: LakeFSFileNode, // the parent node + val ownerEmail: String, + val size: Option[Long] = None, // size of the file in bytes, None if directory + var children: Option[List[LakeFSFileNode]] = None // Only populated if 'type' is 'directory' +) { + + // Ensure the type is either "file" or "directory" + require(nodeType == "file" || nodeType == "directory", "type must be 'file' or 'directory'") + + // Getters for the parameters + def getName: String = name + + def getNodeType: String = nodeType + + def getParent: LakeFSFileNode = parent + + def getOwnerEmail: String = ownerEmail + + def getSize: Option[Long] = size + + def getChildren: List[LakeFSFileNode] = children.getOrElse(List()) + + // Method to get the full file path + def getFilePath: String = { + val pathComponents = new mutable.ArrayBuffer[String]() + var currentNode: LakeFSFileNode = this + while (currentNode != null) { + if (currentNode.parent != null) { // Skip the root node to avoid double slashes + pathComponents.prepend(currentNode.name) + } + currentNode = currentNode.parent + } + "/" + pathComponents.mkString("/") + } +} + +object LakeFSFileNode { + + /** + * Converts a map of LakeFS committed objects into a structured file node tree. + * + * The tree is rooted at the resource-type segment (`dataset` or `model`), which + * [[LakeFSFileNode.getFilePath]] emits as the first path component. That prefix is + * what `FileResolver` keys on to pick the backing table, so it must match the + * resource the objects actually came from. + * + * @param resourceType The resource type the objects belong to (dataset or model). + * @param map A mapping from `(ownerEmail, resourceName, versionName)` to a list of committed objects. + * @return A list of root-level file nodes. + */ + def fromLakeFSRepositoryCommittedObjects( + resourceType: ResourceType.Value, + map: Map[(String, String, String), List[ObjectStats]] + ): List[LakeFSFileNode] = { + val rootNode = new LakeFSFileNode("/", "directory", null, "") + + // Root the tree at the resource-type prefix node (a directory node named e.g. "dataset"). + val resourceTypeNode = + new LakeFSFileNode(resourceType.toString, "directory", rootNode, "") + rootNode.children = Some(List(resourceTypeNode)) + + // Owner level nodes map + val ownerNodes = mutable.Map[String, LakeFSFileNode]() + + map.foreach { + case ((ownerEmail, resourceName, versionName), objects) => + val ownerNode = ownerNodes.getOrElseUpdate( + ownerEmail, { + val newNode = new LakeFSFileNode(ownerEmail, "directory", resourceTypeNode, ownerEmail) + resourceTypeNode.children = Some(resourceTypeNode.getChildren :+ newNode) + newNode + } + ) + + val resourceNode = ownerNode.getChildren.find(_.getName == resourceName).getOrElse { + val newNode = new LakeFSFileNode(resourceName, "directory", ownerNode, ownerEmail) + ownerNode.children = Some(ownerNode.getChildren :+ newNode) + newNode + } + + val versionNode = resourceNode.getChildren.find(_.getName == versionName).getOrElse { + val newNode = new LakeFSFileNode(versionName, "directory", resourceNode, ownerEmail) + resourceNode.children = Some(resourceNode.getChildren :+ newNode) + newNode + } + + // Every node for this version, keyed by its path relative to the version root. + // Leaves are registered too, so a name used as both object and directory is caught. + val nodeMap = mutable.Map[String, LakeFSFileNode]() + nodeMap("") = versionNode // Root of the resource version + + // Process each object (file or directory) from LakeFS + objects.foreach { obj => + val pathParts = obj.getPath.split("/").toList + var currentPath = "" + var parentNode: LakeFSFileNode = versionNode + + pathParts.zipWithIndex.foreach { + case (part, idx) => + currentPath = if (currentPath.isEmpty) part else s"$currentPath/$part" + + // Positional, not by value: a path that repeats its final segment (e.g. + // "model/model") would otherwise treat the intermediate directory as the leaf, + // giving it the object's size and nesting the real file underneath it. + val isFile = idx == pathParts.length - 1 + val nodeType = if (isFile) "file" else "directory" + val fileSize = if (isFile) Some(obj.getSizeBytes.longValue()) else None + + val node = nodeMap.get(currentPath) match { + // LakeFS allows both "model" and "model/weights.bin"; a tree keeps one, and the directory has to win to hold the deeper object. + case Some(existing) if !isFile && existing.getNodeType == "file" => + val promoted = new LakeFSFileNode(part, "directory", parentNode, ownerEmail) + parentNode.children = Some( + parentNode.getChildren.map(child => if (child eq existing) promoted else child) + ) + nodeMap(currentPath) = promoted + promoted Review Comment: For now, we can keep this, lets just leave TODO comment -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
