Function validateMetaDBFilesTable

integration/metadb_test.go:397–484

(t *testing.T, db *database.DB, directory string, dbName string, importID util.FixedString)

}

func validateMetaDBFilesTable(t *testing.T, db *database.DB, directory string, dbName string, importID util.FixedString) {
	t.Helper()

	// set the context
	ctx := clickhouse.Context(db.GetContext(), clickhouse.WithParameters(clickhouse.Parameters{
		"table":    "files",
		"database": dbName,
		"importID": importID.Hex(),
	}))

	// files table exists
	t.Run("Table Exists", func(t *testing.T) {
		var exists uint8
		err := db.Conn.QueryRow(ctx, `EXISTS TABLE {table:Identifier}`).Scan(&exists)
		require.NoError(t, err)
		require.EqualValues(t, 1, exists, "files table must exist")
	})

	// verify that files entries for import are correct
	t.Run("Files List Correct", func(t *testing.T) {

		// get the files listed in the files table for the first import
		rows, err := db.Conn.Query(ctx, `
				SELECT path
				FROM {table:Identifier}
				WHERE database = {database:String} AND import_id = unhex({importID:String})
			`)
		require.NoError(t, err)
		defer rows.Close()

		// type fileVersions

		// map to track the versions of each file
		dbFiles := make(map[string]struct {
			log   bool
			logGz bool
		})

		for rows.Next() {
			var result struct {
				Path string `ch:"path"`
			}
			err = rows.ScanStruct(&result)
			require.NoError(t, err)

			// get the filename without the path
			name := fp.Base(result.Path)

			// set map key as the filename without the .gz extension
			normalizedFilename := strings.TrimSuffix(name, ".gz")

			// get the version info for the file
			versionInfo := dbFiles[normalizedFilename]

			// set the version info for the file
			if strings.HasSuffix(name, ".gz") {
				versionInfo.logGz = true
			} else {