<commit_before>#
# Copyright (c) 2004-2006 rPath, Inc.
#
# This program is distributed under the terms of the Common Public License,
# version 1.0. A copy of this license should have been distributed with this
# source file in a file called LICENSE. If it is not present, the license
# is always available at http://www.opensource.org/licenses/cpl.php.
#
# This program is distributed in the hope that it will be useful, but
# without any warranty; without even the implied warranty of merchantability
# or fitness for a particular purpose. See the Common Public License for
# full details.
#

import itertools

from conary import files, metadata, trove, versions, changelog
from conary.deps import deps
from conary.lib import util, tracelog
from conary.local import deptable
from conary.local import versiontable
from conary.repository import errors
from conary.repository.netrepos import instances, items, keytable, flavors
from conary.repository.netrepos import troveinfo, versionops, cltable
from conary.server import schema

class LocalRepVersionTable(versiontable.VersionTable):

    def getId(self, theId, itemId):
        cu = self.db.cursor()
        cu.execute("""
        SELECT Versions.version, Nodes.timeStamps
        FROM Nodes JOIN Versions USING (versionId)
        WHERE Nodes.versionId=? AND Nodes.itemId=?""",
		   theId, itemId)
	try:
	    (s, t) = cu.next()
	    v = self._makeVersion(s, t)
	    return v
	except StopIteration:
            raise KeyError, theId

    def getTimeStamps(self, version, itemId):
        cu = self.db.cursor()
        cu.execute("""
        SELECT timeStamps FROM Nodes
        WHERE versionId = (SELECT versionId from Versions WHERE version=?)
          AND itemId=?""", version.asString(), itemId)
	try:
	    (t,) = cu.next()
	    return [ float(x) for x in t.split(":") ]
	except StopIteration:
            raise KeyError, itemId

class TroveStore:
    def __init__(self, db, log = None):
	self.db = db

	self.items = items.Items(self.db)
	self.flavors = flavors.Flavors(self.db)
        self.branchTable = versionops.BranchTable(self.db)
        self.changeLogs = cltable.ChangeLogTable(self.db)

	self.versionTable = LocalRepVersionTable(self.db)
	self.versionOps = versionops.SqlVersioning(
            self.db, self.versionTable, self.branchTable)
	self.instances = instances.InstanceTable(self.db)

        self.keyTable = keytable.OpenPGPKeyTable(self.db)
        self.depTables = deptable.DependencyTables(self.db)
        self.metadataTable = metadata.MetadataTable(self.db, create = False)
        self.troveInfoTable = troveinfo.TroveInfoTable(self.db)

        self.streamIdCache = {}
	self.needsCleanup = False
        self.log = log or tracelog.getLog(None)

    def __del__(self):
        self.db = self.log = None

    def getLabelId(self, label):
        self.versionOps.labels.getOrAddId(label)

    def getItemId(self, item):
        return self.items.getOrAddId(item)

    def getInstanceId(self, itemId, versionId, flavorId, clonedFromId,
                      isRedirect, isPresent = True):
 	theId = self.instances.get((itemId, versionId, flavorId), None)
	if theId == None:
	    theId = self.instances.addId(itemId, versionId, flavorId,
                                         clonedFromId,
					 isRedirect, isPresent = isPresent)
        # XXX we shouldn't have to do this unconditionally
        if isPresent:
	    self.instances.setPresent(theId, 1)
            self.items.setTroveFlag(itemId, 1)
 	return theId

    def getVersionId(self, version, cache):
	theId = cache.get(version, None)
	if theId:
	    return theId

	theId = self.versionTable.get(version, None)
	if theId == None:
	    theId = self.versionTable.addId(version)

	cache[version] = theId

	return theId

    def getFullVersion(self, item, version):
	"""
	Updates version with full timestamp information.
	"""
	cu = self.db.cursor()
	cu.execute("""
	    SELECT timeStamps FROM Nodes WHERE
		itemId=(SELECT itemId FROM Items WHERE item=?) AND
		versionId=(SELECT versionId FROM Versions WHERE version=?)
	""", item, version.asString())

	timeStamps = cu.fetchone()[0]
	version.setTimeStamps([float(x) for x in timeStamps.split(":")])

    def createTroveBranch(self, troveName, branch):
	itemId = self.getItemId(troveName)
	branchId = self.versionOps.createBranch(itemId, branch)

    def troveLatestVersion(self, troveName, branch):
	"""
	Returns None if no versions of troveName exist on the branch.
	"""
	cu = self.db.cursor()
	cu.execute("""
	    SELECT version, timeStamps FROM
		(SELECT itemId AS AitemId, branchId as AbranchId FROM labelMap
		    WHERE itemId=(SELECT itemId from Items
				WHERE item=?)
		    AND branchId=(SELECT branchId FROM Branches
				WHERE branch=?)
		) INNER JOIN Latest ON
		    AitemId=Latest.itemId AND AbranchId=Latest.branchId
		INNER JOIN Nodes ON
		    AitemId=Nodes.itemId AND Latest.versionId=Nodes.versionId
		INNER JOIN Versions ON
		    Nodes.versionId = versions.versionId
                ORDER BY
                    Nodes.finalTimeStamp
                LIMIT 1
	""", troveName, branch.asString())
        try:
	    (verStr, timeStamps) = cu.next()
            return versions.VersionFromString(verStr,
		    timeStamps = [ float(x) for x in timeStamps.split(":") ] )
        except StopIteration:
            raise KeyError, (troveName, branch)

    def getTroveFlavors(self, troveDict):
	cu = self.db.cursor()
	vMap = {}
	outD = {}
	# I think we might be better of intersecting subqueries rather
	# then using all of the and's in this join

        schema.resetTable(cu, 'itf')

        for troveName in troveDict.keys():
            outD[troveName] = {}
            for version in troveDict[troveName]:
                outD[troveName][version] = []
                versionStr = version.asString()
                vMap[versionStr] = version
                cu.execute("INSERT INTO itf VALUES (?, ?, ?)",
                           (troveName, versionStr, versionStr),
                           start_transaction = False)

        cu.execute("""
            SELECT aItem, fullVersion, Flavors.flavor FROM
                (SELECT Items.itemId AS aItemId,
                        versions.versionId AS aVersionId,
                        Items.item AS aItem,
                        fullVersion FROM
                    itf INNER JOIN Items ON itf.item = Items.item
                        INNER JOIN versions ON itf.version = versions.version) as ItemVersions
                INNER JOIN instances ON
                    aItemId = instances.itemId AND
                    aVersionId = instances.versionId
                INNER JOIN flavors ON
                    instances.flavorId = flavors.flavorId
                ORDER BY aItem, fullVersion
        """)

        for (item, verString, flavor) in cu:
            ver = vMap[verString]
            outD[item][ver].append(flavor)

	return outD

    def iterTroveNames(self):
        cu = self.db.cursor()
        cu.execute("SELECT DISTINCT Items.item as item "
                   " FROM Instances JOIN Items USING(itemId) "
                   " WHERE Instances.isPresent=1 ORDER BY item");

        for (item,) in cu:
            yield item

    def addTrove(self, trv):
	cu = self.db.cursor()

        schema.resetTable(cu, 'NewFiles')
        schema.resetTable(cu, 'NeededFlavors')

	self.fileVersionCache = {}
	return (cu, trv)

    def addTroveDone(self, troveInfo):
	versionCache = {}
	(cu, trv) = troveInfo

        self.log(3, trv)

	troveVersion = trv.getVersion()
	troveItemId = self.getItemId(trv.getName())
        sourceName = trv.troveInfo.sourceName()

        # Pull out the clonedFromId
        clonedFrom = trv.troveInfo.clonedFrom()
        clonedFromId = None
        if clonedFrom:
            clonedFromId = self.versionTable.get(clonedFrom, None)
            if clonedFromId is None:
                clonedFromId = self.versionTable.addId(clonedFrom)

        isPackage = (not trv.getName().startswith('group') and
                     not trv.getName().startswith('fileset') and
                     ':' not in trv.getName())

	# does this version already exist (for another flavor?)
	newVersion = False
	troveVersionId = self.versionTable.get(troveVersion, None)
	if troveVersionId is not None:
	    nodeId = self.versionOps.nodes.getRow(troveItemId,
						  troveVersionId, None)

	troveFlavor = trv.getFlavor()

	# start off by creating the flavors we need; we could combine this
	# to some extent with the file table creation below, but there are
	# normally very few flavors per trove so this probably better
	flavorsNeeded = {}
	if troveFlavor is not None:
	    flavorsNeeded[troveFlavor] = True

	for (name, version, flavor) in trv.iterTroveList(strongRefs = True,
                                                           weakRefs = True):
	    if flavor is not None:
		flavorsNeeded[flavor] = True

        for (name, branch, flavor) in trv.iterRedirects():
            if flavor is not None:
                flavorsNeeded[flavor] = True

	flavorIndex = {}
	for flavor in flavorsNeeded.iterkeys():
	    flavorIndex[flavor.freeze()] = flavor
	    cu.execute("INSERT INTO NeededFlavors VALUES(?)", flavor.freeze())

	del flavorsNeeded

	# it seems like there must be a better way to do this, but I can't
	# figure it out. I *think* inserting into a view would help, but I
	# can't with sqlite.

	cu.execute("""SELECT NeededFlavors.flavor FROM
			NeededFlavors LEFT OUTER JOIN Flavors ON
			    NeededFlavors.flavor = Flavors.Flavor
			WHERE Flavors.flavorId is NULL""")
        # make a list of the flavors we're going to create.  Add them
        # after we have retreived all of the rows from this select
        l = []
	for (flavorStr,) in cu:
            l.append(flavorIndex[flavorStr])
        for flavor in l:
	    self.flavors.createFlavor(flavor)

	flavors = {}
	cu.execute("""SELECT Flavors.flavor, Flavors.flavorId FROM
			NeededFlavors INNER JOIN Flavors ON
			NeededFlavors.flavor = Flavors.flavor""")
	for (flavorStr, flavorId) in cu:
	    flavors[flavorIndex[flavorStr]] = flavorId

	del flavorIndex

	if troveFlavor is not None:
	    troveFlavorId = flavors[troveFlavor]
	else:
	    troveFlavorId = 0

	if troveVersionId is None or nodeId is None:
	    (nodeId, troveVersionId) = self.versionOps.createVersion(
                troveItemId, troveVersion, troveFlavorId, sourceName)
	    newVersion = True

	    if trv.getChangeLog() and trv.getChangeLog().getName():
		self.changeLogs.add(nodeId, trv.getChangeLog())
            updateLatest = False
        else:
            updateLatest = True

	# the instance may already exist (it could be referenced by a package
	# which has already been added)
	troveInstanceId = self.getInstanceId(troveItemId, troveVersionId,
					     troveFlavorId, clonedFromId,
                                             trv.isRedirect(),
                                             isPresent = True)
        assert(cu.execute("SELECT COUNT(*) from TroveTroves WHERE "
                          "instanceId=?", troveInstanceId).next()[0] == 0)

        if updateLatest:
            # this name/version already exists, so this must be a new
            # flavor. update the latest table as needed
            troveBranchId = self.branchTable[troveVersion.branch()]
            cu.execute("""
            DELETE FROM Latest WHERE branchId=? AND itemId=? AND flavorId=?
            """, troveBranchId, troveItemId, troveFlavorId)
            cu.execute("""
            INSERT INTO Latest
                (itemId, branchId, flavorId, versionId)
            SELECT %d, %d, %d, Instances.versionId
            FROM
                Instances JOIN Nodes USING(itemId, versionId)
            WHERE
                Instances.itemId=?
            AND Instances.flavorId=?
            AND Nodes.branchId=?
            ORDER BY finalTimestamp DESC
            LIMIT 1
            """ %(troveItemId, troveBranchId, troveFlavorId),
                       (troveItemId, troveFlavorId, troveBranchId))

        self.depTables.add(cu, trv, troveInstanceId)

        # Fold NewFiles into FileStreams
        #
        # NewFiles can contain duplicate entries for the same fileId,
        # (with stream being NULL or not), so the FileStreams update
        # is happening in three steps:
        # 1. Update existing fileIds (while avoiding  a full table scan)
        # 2. Insert new fileIds with non-NULL streams
        # 3. Insert new fileIds that might have NULL streams.

        # Note: writing the next two steps in a single query causes
        # very slow full table scans on FileStreams. Don't get fancy.

        # get the common entries we're gonna update
        cu2 = self.db.cursor()
        cu2.execute("""
        SELECT NewFiles.fileId, NewFiles.stream
        FROM NewFiles
        JOIN FileStreams USING(fileId)
        WHERE FileStreams.stream IS NULL
        AND NewFiles.stream IS NOT NULL
        """)
        # Note: PostgreSQL and MySQL have support for non-SQL standard
        # multi-table updates that we could use to do this in one step.
        # This two step should work on everything though --gafton
        for (fileId, stream) in cu2:
            cu.execute("UPDATE FileStreams SET stream = ? "
                       "WHERE fileId = ?", (stream, fileId))
        del cu2

        # select the new non-NULL streams out of NewFiles and Insert them in FileStreams
        cu.execute("""
        INSERT INTO FileStreams (fileId, stream)
        SELECT DISTINCT NF.fileId, NF.stream
        FROM NewFiles AS NF
        LEFT OUTER JOIN FileStreams AS FS USING(fileId)
        WHERE FS.fileId IS NULL
          AND NF.stream IS NOT NULL
        """)<commit_msg>deadlock avoidance requires the use of fetchall()<commit_after>#
# Copyright (c) 2004-2006 rPath, Inc.
#
# This program is distributed under the terms of the Common Public License,
# version 1.0. A copy of this license should have been distributed with this
# source file in a file called LICENSE. If it is not present, the license
# is always available at http://www.opensource.org/licenses/cpl.php.
#
# This program is distributed in the hope that it will be useful, but
# without any warranty; without even the implied warranty of merchantability
# or fitness for a particular purpose. See the Common Public License for
# full details.
#

import itertools

from conary import files, metadata, trove, versions, changelog
from conary.deps import deps
from conary.lib import util, tracelog
from conary.local import deptable
from conary.local import versiontable
from conary.repository import errors
from conary.repository.netrepos import instances, items, keytable, flavors
from conary.repository.netrepos import troveinfo, versionops, cltable
from conary.server import schema

class LocalRepVersionTable(versiontable.VersionTable):

    def getId(self, theId, itemId):
        cu = self.db.cursor()
        cu.execute("""
        SELECT Versions.version, Nodes.timeStamps
        FROM Nodes JOIN Versions USING (versionId)
        WHERE Nodes.versionId=? AND Nodes.itemId=?""",
		   theId, itemId)
	try:
	    (s, t) = cu.next()
	    v = self._makeVersion(s, t)
	    return v
	except StopIteration:
            raise KeyError, theId

    def getTimeStamps(self, version, itemId):
        cu = self.db.cursor()
        cu.execute("""
        SELECT timeStamps FROM Nodes
        WHERE versionId = (SELECT versionId from Versions WHERE version=?)
          AND itemId=?""", version.asString(), itemId)
	try:
	    (t,) = cu.next()
	    return [ float(x) for x in t.split(":") ]
	except StopIteration:
            raise KeyError, itemId

class TroveStore:
    def __init__(self, db, log = None):
	self.db = db

	self.items = items.Items(self.db)
	self.flavors = flavors.Flavors(self.db)
        self.branchTable = versionops.BranchTable(self.db)
        self.changeLogs = cltable.ChangeLogTable(self.db)

	self.versionTable = LocalRepVersionTable(self.db)
	self.versionOps = versionops.SqlVersioning(
            self.db, self.versionTable, self.branchTable)
	self.instances = instances.InstanceTable(self.db)

        self.keyTable = keytable.OpenPGPKeyTable(self.db)
        self.depTables = deptable.DependencyTables(self.db)
        self.metadataTable = metadata.MetadataTable(self.db, create = False)
        self.troveInfoTable = troveinfo.TroveInfoTable(self.db)

        self.streamIdCache = {}
	self.needsCleanup = False
        self.log = log or tracelog.getLog(None)

    def __del__(self):
        self.db = self.log = None

    def getLabelId(self, label):
        self.versionOps.labels.getOrAddId(label)

    def getItemId(self, item):
        return self.items.getOrAddId(item)

    def getInstanceId(self, itemId, versionId, flavorId, clonedFromId,
                      isRedirect, isPresent = True):
 	theId = self.instances.get((itemId, versionId, flavorId), None)
	if theId == None:
	    theId = self.instances.addId(itemId, versionId, flavorId,
                                         clonedFromId,
					 isRedirect, isPresent = isPresent)
        # XXX we shouldn't have to do this unconditionally
        if isPresent:
	    self.instances.setPresent(theId, 1)
            self.items.setTroveFlag(itemId, 1)
 	return theId

    def getVersionId(self, version, cache):
	theId = cache.get(version, None)
	if theId:
	    return theId

	theId = self.versionTable.get(version, None)
	if theId == None:
	    theId = self.versionTable.addId(version)

	cache[version] = theId

	return theId

    def getFullVersion(self, item, version):
	"""
	Updates version with full timestamp information.
	"""
	cu = self.db.cursor()
	cu.execute("""
	    SELECT timeStamps FROM Nodes WHERE
		itemId=(SELECT itemId FROM Items WHERE item=?) AND
		versionId=(SELECT versionId FROM Versions WHERE version=?)
	""", item, version.asString())

	timeStamps = cu.fetchone()[0]
	version.setTimeStamps([float(x) for x in timeStamps.split(":")])

    def createTroveBranch(self, troveName, branch):
	itemId = self.getItemId(troveName)
	branchId = self.versionOps.createBranch(itemId, branch)

    def troveLatestVersion(self, troveName, branch):
	"""
	Returns None if no versions of troveName exist on the branch.
	"""
	cu = self.db.cursor()
	cu.execute("""
	    SELECT version, timeStamps FROM
		(SELECT itemId AS AitemId, branchId as AbranchId FROM labelMap
		    WHERE itemId=(SELECT itemId from Items
				WHERE item=?)
		    AND branchId=(SELECT branchId FROM Branches
				WHERE branch=?)
		) INNER JOIN Latest ON
		    AitemId=Latest.itemId AND AbranchId=Latest.branchId
		INNER JOIN Nodes ON
		    AitemId=Nodes.itemId AND Latest.versionId=Nodes.versionId
		INNER JOIN Versions ON
		    Nodes.versionId = versions.versionId
                ORDER BY
                    Nodes.finalTimeStamp
                LIMIT 1
	""", troveName, branch.asString())
        try:
	    (verStr, timeStamps) = cu.next()
            return versions.VersionFromString(verStr,
		    timeStamps = [ float(x) for x in timeStamps.split(":") ] )
        except StopIteration:
            raise KeyError, (troveName, branch)

    def getTroveFlavors(self, troveDict):
	cu = self.db.cursor()
	vMap = {}
	outD = {}
	# I think we might be better of intersecting subqueries rather
	# then using all of the and's in this join

        schema.resetTable(cu, 'itf')

        for troveName in troveDict.keys():
            outD[troveName] = {}
            for version in troveDict[troveName]:
                outD[troveName][version] = []
                versionStr = version.asString()
                vMap[versionStr] = version
                cu.execute("INSERT INTO itf VALUES (?, ?, ?)",
                           (troveName, versionStr, versionStr),
                           start_transaction = False)

        cu.execute("""
            SELECT aItem, fullVersion, Flavors.flavor FROM
                (SELECT Items.itemId AS aItemId,
                        versions.versionId AS aVersionId,
                        Items.item AS aItem,
                        fullVersion FROM
                    itf INNER JOIN Items ON itf.item = Items.item
                        INNER JOIN versions ON itf.version = versions.version) as ItemVersions
                INNER JOIN instances ON
                    aItemId = instances.itemId AND
                    aVersionId = instances.versionId
                INNER JOIN flavors ON
                    instances.flavorId = flavors.flavorId
                ORDER BY aItem, fullVersion
        """)

        for (item, verString, flavor) in cu:
            ver = vMap[verString]
            outD[item][ver].append(flavor)

	return outD

    def iterTroveNames(self):
        cu = self.db.cursor()
        cu.execute("SELECT DISTINCT Items.item as item "
                   " FROM Instances JOIN Items USING(itemId) "
                   " WHERE Instances.isPresent=1 ORDER BY item");

        for (item,) in cu:
            yield item

    def addTrove(self, trv):
	cu = self.db.cursor()

        schema.resetTable(cu, 'NewFiles')
        schema.resetTable(cu, 'NeededFlavors')

	self.fileVersionCache = {}
	return (cu, trv)

    def addTroveDone(self, troveInfo):
	versionCache = {}
	(cu, trv) = troveInfo

        self.log(3, trv)

	troveVersion = trv.getVersion()
	troveItemId = self.getItemId(trv.getName())
        sourceName = trv.troveInfo.sourceName()

        # Pull out the clonedFromId
        clonedFrom = trv.troveInfo.clonedFrom()
        clonedFromId = None
        if clonedFrom:
            clonedFromId = self.versionTable.get(clonedFrom, None)
            if clonedFromId is None:
                clonedFromId = self.versionTable.addId(clonedFrom)

        isPackage = (not trv.getName().startswith('group') and
                     not trv.getName().startswith('fileset') and
                     ':' not in trv.getName())

	# does this version already exist (for another flavor?)
	newVersion = False
	troveVersionId = self.versionTable.get(troveVersion, None)
	if troveVersionId is not None:
	    nodeId = self.versionOps.nodes.getRow(troveItemId,
						  troveVersionId, None)

	troveFlavor = trv.getFlavor()

	# start off by creating the flavors we need; we could combine this
	# to some extent with the file table creation below, but there are
	# normally very few flavors per trove so this probably better
	flavorsNeeded = {}
	if troveFlavor is not None:
	    flavorsNeeded[troveFlavor] = True

	for (name, version, flavor) in trv.iterTroveList(strongRefs = True,
                                                           weakRefs = True):
	    if flavor is not None:
		flavorsNeeded[flavor] = True

        for (name, branch, flavor) in trv.iterRedirects():
            if flavor is not None:
                flavorsNeeded[flavor] = True

	flavorIndex = {}
	for flavor in flavorsNeeded.iterkeys():
	    flavorIndex[flavor.freeze()] = flavor
	    cu.execute("INSERT INTO NeededFlavors VALUES(?)", flavor.freeze())

	del flavorsNeeded

	# it seems like there must be a better way to do this, but I can't
	# figure it out. I *think* inserting into a view would help, but I
	# can't with sqlite.

	cu.execute("""SELECT NeededFlavors.flavor FROM
			NeededFlavors LEFT OUTER JOIN Flavors ON
			    NeededFlavors.flavor = Flavors.Flavor
			WHERE Flavors.flavorId is NULL""")
        # make a list of the flavors we're going to create.  Add them
        # after we have retreived all of the rows from this select
        l = []
	for (flavorStr,) in cu:
            l.append(flavorIndex[flavorStr])
        for flavor in l:
	    self.flavors.createFlavor(flavor)

	flavors = {}
	cu.execute("""SELECT Flavors.flavor, Flavors.flavorId FROM
			NeededFlavors INNER JOIN Flavors ON
			NeededFlavors.flavor = Flavors.flavor""")
	for (flavorStr, flavorId) in cu:
	    flavors[flavorIndex[flavorStr]] = flavorId

	del flavorIndex

	if troveFlavor is not None:
	    troveFlavorId = flavors[troveFlavor]
	else:
	    troveFlavorId = 0

	if troveVersionId is None or nodeId is None:
	    (nodeId, troveVersionId) = self.versionOps.createVersion(
                troveItemId, troveVersion, troveFlavorId, sourceName)
	    newVersion = True

	    if trv.getChangeLog() and trv.getChangeLog().getName():
		self.changeLogs.add(nodeId, trv.getChangeLog())
            updateLatest = False
        else:
            updateLatest = True

	# the instance may already exist (it could be referenced by a package
	# which has already been added)
	troveInstanceId = self.getInstanceId(troveItemId, troveVersionId,
					     troveFlavorId, clonedFromId,
                                             trv.isRedirect(),
                                             isPresent = True)
        assert(cu.execute("SELECT COUNT(*) from TroveTroves WHERE "
                          "instanceId=?", troveInstanceId).next()[0] == 0)

        if updateLatest:
            # this name/version already exists, so this must be a new
            # flavor. update the latest table as needed
            troveBranchId = self.branchTable[troveVersion.branch()]
            cu.execute("""
            DELETE FROM Latest WHERE branchId=? AND itemId=? AND flavorId=?
            """, troveBranchId, troveItemId, troveFlavorId)
            cu.execute("""
            INSERT INTO Latest
                (itemId, branchId, flavorId, versionId)
            SELECT %d, %d, %d, Instances.versionId
            FROM
                Instances JOIN Nodes USING(itemId, versionId)
            WHERE
                Instances.itemId=?
            AND Instances.flavorId=?
            AND Nodes.branchId=?
            ORDER BY finalTimestamp DESC
            LIMIT 1
            """ %(troveItemId, troveBranchId, troveFlavorId),
                       (troveItemId, troveFlavorId, troveBranchId))

        self.depTables.add(cu, trv, troveInstanceId)

        # Fold NewFiles into FileStreams
        #
        # NewFiles can contain duplicate entries for the same fileId,
        # (with stream being NULL or not), so the FileStreams update
        # is happening in three steps:
        # 1. Update existing fileIds (while avoiding  a full table scan)
        # 2. Insert new fileIds with non-NULL streams
        # 3. Insert new fileIds that might have NULL streams.

        # Note: writing the next two steps in a single query causes
        # very slow full table scans on FileStreams. Don't get fancy.

        # get the common entries we're gonna update. In the extreme
        # case of binary shadowing this might require a bit of of
        # memory for large troves, but it is preferable to constant
        # full table scans in the much more common cases
        cu.execute("""
        SELECT NewFiles.fileId, NewFiles.stream
        FROM NewFiles
        JOIN FileStreams USING(fileId)
        WHERE FileStreams.stream IS NULL
        AND NewFiles.stream IS NOT NULL
        """)
        # Note: PostgreSQL and MySQL have support for non-SQL standard
        # multi-table updates that we could use to do this in one step.
        # This two step should work on everything though --gafton
        for (fileId, stream) in cu.fetchall():
            cu.execute("UPDATE FileStreams SET stream = ? "
                       "WHERE fileId = ?", (stream, fileId))

        # select the new non-NULL streams out of NewFiles and Insert
        # them in FileStreams
        cu.execute("""
        INSERT INTO FileStreams (fileId, stream)
        SELECT DISTINCT NF.fileId, NF.stream
        FROM NewFiles AS NF
        LEFT OUTER JOIN FileStreams AS FS USING(fileId)
        WHERE FS.fileId IS NULL
          AND NF.stream IS NOT NULL
        """)
{'fraction_non_alphanumeric': 0.08560564910374796, 'fraction_numerical': 0.001810610175629187, 'mean_word_length': 2.393879808283151, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3734483', 'n_tokens_mistral': 8579, 'n_tokens_neox': 7983, 'n_words': 2459}